Repository: etcd-io/etcd
File: tests/integration/v3_lease_test.go (Go)

Code before:
		require.Eventually(t, func() bool {
			return getMetricVal(follower, "Unavailable") == prevUnavailableCount+1
		}, 3*time.Second, 100*time.Millisecond)

		// We didn't see the Canceled count increase
		require.Equal(t, prevCanceledCount, getMetricVal(follower, "Canceled"))
	})

	t.Run("forwarding times out", func(t *testing.T) {
		follower, _, keepAliveClient, leaseID := setupLeaseKeepAliveForwarding(t)
		defer keepAliveClient.CloseSend()

		prevUnavailableCount := getMetricVal(follower, "Unavailable")
		require.NoError(t, keepAliveClient.Send(&pb.LeaseKeepAliveRequest{ID: leaseID}))

		_, err := keepAliveClient.Recv()
		require.Equal(t, rpctypes.ErrGRPCTimeout, err)

		// We should observe Unavailable metric went up by 1
		require.Eventually(t, func() bool {
			return getMetricVal(follower, "Unavailable") == prevUnavailableCount+1
		}, 3*time.Second, 100*time.Millisecond)
	})
}

func setupLeaseKeepAliveForwarding(t *testing.T) (*integration.Member, context.CancelFunc, pb.Lease_LeaseKeepAliveClient, int64) {
	t.Helper()
	cluster := integration.NewCluster(t, &integration.ClusterConfig{Size: 3})
	t.Cleanup(func() { cluster.Terminate(t) })

	leaderIdx := cluster.WaitLeader(t)
	followerIdx := (leaderIdx + 1) % 3
	follower := cluster.Members[followerIdx]

	// Create a lease
	grantResp, err := integration.ToGRPC(cluster.Client(leaderIdx)).Lease.LeaseGrant(
		t.Context(),
		&pb.LeaseGrantRequest{TTL: 30},
	)
	require.NoError(t, err)
	require.Empty(t, grantResp.Error)
	leaseID := grantResp.ID

	// Start keepalive stream on follower so that it will forward request to leader via HTTP
	ctx, cancel := context.WithCancel(t.Context())
	leaseClient := integration.ToGRPC(follower.Client).Lease
	keepAliveClient, err := leaseClient.LeaseKeepAlive(ctx)
	require.NoError(t, err)

	// Make sure the stream is actually functioning first.
	require.NoError(t, keepAliveClient.Send(&pb.LeaseKeepAliveRequest{ID: leaseID}))

Review comment (suggestion):
Please remove the comments, or at least replace them with logs. Commenting just WHAT? is happening is useless, as what matters is code not comment. If you want to comment you should focus on WHY? to explain things not expressed with code.

Code after:
		// after cancel() regardless of the actual server response.
		prevCanceledCount := getLeaseKeepAliveMetric(t, follower, "Canceled")
		prevUnavailableCount := getLeaseKeepAliveMetric(t, follower, "Unavailable")

		require.NoError(t, keepAliveClient.Send(&pb.LeaseKeepAliveRequest{ID: leaseID}))
		time.Sleep(50 * time.Millisecond)
		cancel()

		// Client sees Canceled (gRPC returns this immediately after cancel()),
		// but server actually generated Unavailable (verified by metrics below).
		_, err = keepAliveClient.Recv()
		require.Equal(t, codes.Canceled, status.Code(err))

		require.Eventually(t, func() bool {
			return getLeaseKeepAliveMetric(t, follower, "Unavailable") == prevUnavailableCount+1
		}, 3*time.Second, 100*time.Millisecond)
		require.Equal(t, prevCanceledCount, getLeaseKeepAliveMetric(t, follower, "Canceled"))
	})

	t.Run("forwarding times out", func(t *testing.T) {
		leaderClient, follower := setupLeaseForwardingCluster(t)

		grantResp, err := leaderClient.LeaseGrant(t.Context(), &pb.LeaseGrantRequest{TTL: 30})
		require.NoError(t, err)
		leaseID := grantResp.ID

		keepAliveClient, err := integration.ToGRPC(follower.Client).Lease.LeaseKeepAlive(t.Context())
		require.NoError(t, err)
		defer keepAliveClient.CloseSend()

		require.NoError(t, keepAliveClient.Send(&pb.LeaseKeepAliveRequest{ID: leaseID}))
		_, err = keepAliveClient.Recv()
		require.NoError(t, err)

		sleepBeforeServingLeaseRenew(t, sleepDuration)

		prevUnavailableCount := getLeaseKeepAliveMetric(t, follower, "Unavailable")
		require.NoError(t, keepAliveClient.Send(&pb.LeaseKeepAliveRequest{ID: leaseID}))

		_, err = keepAliveClient.Recv()
		require.Equal(t, rpctypes.ErrGRPCTimeout, err)

		require.Eventually(t, func() bool {
			return getLeaseKeepAliveMetric(t, follower, "Unavailable") == prevUnavailableCount+1
		}, 3*time.Second, 100*time.Millisecond)
	})
}

func setupLeaseForwardingCluster(t *testing.T) (pb.LeaseClient, *integration.Member) {
	t.Helper()
	cluster := integration.NewCluster(t, &integration.ClusterConfig{Size: 3})