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

Code before:
		e2e.WithSnapshotCatchUpEntries(uint64(snapCount)),
		e2e.WithClusterSize(3),
		e2e.WithKeepDataDir(true),
		e2e.WithGoFailEnabled(true),
		e2e.WithLogLevel("debug"),
		e2e.WithRollingStart(true),
		e2e.WithInitialLeaderIndex(0),
	)
	epc, err := e2e.NewEtcdProcessCluster(t.Context(), t, e2e.WithConfig(cfg))
	require.NoError(t, err)
	defer func() {
		require.NoError(t, epc.Close())
	}()

	t.Log("Step 1: Ensure that each member has about 1 MiB boltdb file.")
	for i := 0; i < 1024; i++ {
		require.NoError(t, epc.Procs[0].Etcdctl().Put(ctx,
			fmt.Sprintf("foo%d", i),
			strings.Repeat("Oops", 1024),
			config.PutOptions{}))
	}

	t.Log(`Step 2: Currently, it's unable to use PeerProxy to block specific\n
data path from one follower to leader. So here, I use SIGSTOP to pause the third member.`)
	require.NoError(t, epc.Procs[2].Failpoints().SetupHTTP(ctx, "applyAfterOpenSnapshot", `sleep("15s")`))
	epc.Procs[2].(*e2e.EtcdServerProcess).Pause()

	t.Log("Step 3: Delete some key values.")
	for i := 0; i < snapCount+20; i++ {
		_, err = epc.Procs[0].Etcdctl().Delete(ctx, fmt.Sprintf("foo%d", i), config.DeleteOptions{})
		require.NoError(t, err)
	}

	t.Log("Step 4: Restarting the first two members to change term and trigger new leader to send snapshot file to the third member.")
	for _, proc := range epc.Procs[:2] {
		require.NoError(t, proc.Restart(ctx))
	}
	epc.Procs[2].(*e2e.EtcdServerProcess).Resume()

	t.Log(`Step 5: After opening snapshot file from new leader, invoke defragment\n
to override boltdb file. So, for the following changes, the third member will commit them into deleted boltdb file.`)
	e2e.AssertProcessLogs(t, epc.Procs[2], "applySnapshot: opened snapshot backend")
	err = epc.Procs[2].Etcdctl().Defragment(ctx, config.DefragOption{Timeout: 30 * time.Second})
	require.NoError(t, err)
	e2e.AssertProcessLogs(t, epc.Procs[2], "applied snapshot")
	require.NoError(t, epc.Procs[2].Failpoints().DeactivateHTTP(ctx, "applyAfterOpenSnapshot"))

	t.Log("Step 6: Transfer leader to the third member.")
	var newLeaderID uint64
	leaderIndex := epc.WaitMembersForLeader(ctx, t, epc.Procs)
	newLeaderID, _, err = getMemberIDByName(ctx, epc.Procs[0].Etcdctl(), epc.Procs[2].Config().Name)

Review comment (suggestion):
After https://github.com/etcd-io/etcd/pull/20553/files#r2321527783 is resolved, then follow suggestion below
```suggestion
	epc.Procs[2].Pause()
```

Code after:
		e2e.WithSnapshotCatchUpEntries(uint64(snapCount)),
		e2e.WithClusterSize(3),
		e2e.WithKeepDataDir(true),
		e2e.WithGoFailEnabled(true),
		e2e.WithLogLevel("debug"),
		e2e.WithRollingStart(true),
		e2e.WithInitialLeaderIndex(0),
	)
	epc, err := e2e.NewEtcdProcessCluster(t.Context(), t, e2e.WithConfig(cfg))
	require.NoError(t, err)
	defer func() {
		require.NoError(t, epc.Close())
	}()

	t.Log("Step 1: Write some data to the cluster")
	for i := 0; i < snapCount*5; i++ {
		require.NoError(t, epc.Procs[0].Etcdctl().Put(ctx,
			fmt.Sprintf("foo%d", i),
			strings.Repeat("Oops", 1024),
			config.PutOptions{}))
	}

	t.Log(`Step 2: Config the third member to sleep 15s after OpenSnapshotBackend and use SIGSTOP to pause it.`)
	require.NoError(t, epc.Procs[2].Failpoints().SetupHTTP(ctx, "applyAfterOpenSnapshot", `sleep("15s")`))
	epc.Procs[2].Pause()

	t.Log("Step 3: Delete some key values to trigger new snapshot on the first two members")
	for i := 0; i < snapCount+20; i++ {
		_, err = epc.Procs[0].Etcdctl().Delete(ctx, fmt.Sprintf("foo%d", i), config.DeleteOptions{})
		require.NoError(t, err)
	}

	t.Log("Step 4: Restarting the first two members to change term and trigger new leader to send snapshot file to the third member.")
	for _, proc := range epc.Procs[:2] {
		require.NoError(t, proc.Restart(ctx))
	}
	epc.Procs[2].Resume()

	t.Log(`Step 5: After opening snapshot file from new leader, invoke defragment\n
to override boltdb file. So, for the following changes, the third member will commit them into deleted boltdb file.`)
	e2e.AssertProcessLogs(t, epc.Procs[2], "applySnapshot: opened snapshot backend")
	err = epc.Procs[2].Etcdctl().Defragment(ctx, config.DefragOption{Timeout: 30 * time.Second})
	require.NoError(t, err)
	e2e.AssertProcessLogs(t, epc.Procs[2], "applied snapshot")
	require.NoError(t, epc.Procs[2].Failpoints().DeactivateHTTP(ctx, "applyAfterOpenSnapshot"))

	t.Log("Step 6: Write some data to the cluster")
	for i := 0; i < snapCount/2; i++ {
		err = epc.Procs[0].Etcdctl().Put(ctx, fmt.Sprintf("foo%d", i), strings.Repeat("Oops", 1), config.PutOptions{})
		require.NoError(t, err)
	}