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

Code before:
		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)
	})

	// This is almost identical to TestV3LeaseRequireLeader, except that it
	// does use WithRequireLeader. In that test, client receives NoLeader error
	// immediately, because monitorLeader() detects leader is missing and thus
	// cancels the server stream with reason ErrGRPCNoLeader.
	//
	// In this test here, monitorLeader() cannot cancel the stream because it's
	// not cancelable without WithRequireLeader. However, we will still receive
	// NoLeader error after the waitLeader() timed out in LeaseRenew().
	t.Run("catches no leader error", func(t *testing.T) {
		leader, follower := setupLeaseForwardingCluster(t)
		leaderClient := integration.ToGRPC(leader.Client).Lease
		followerClient := integration.ToGRPC(follower.Client).Lease

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

		keepAliveClient, err := followerClient.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)
		prevUnavailableCount := getLeaseKeepAliveMetric(t, follower, "Unavailable")


Review comment (suggestion):
Instead of having a comment saying those tests are identical, could you just migrate TestV3LeaseRequireLeader to be a test case here? Having it as separate PR will help showing the current behavior.

Code after:
		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)
	})

	// Client set up with WithRequireLeader() will receive NoLeader error right after
	// monitorLeader() detects leader missing and cancels the server stream with ErrGRPCNoLeader.
	t.Run("catches NoLeader error with WithRequireLeader", func(t *testing.T) {
		leader, follower, anotherFollower := setupLeaseForwardingCluster(t)
		followerClient := integration.ToGRPC(follower.Client).Lease

		prevUnavailableCount := getLeaseKeepAliveMetric(t, follower, "Unavailable")
		leader.Stop(t)
		anotherFollower.Stop(t)

		keepAliveClient, err := followerClient.LeaseKeepAlive(clientv3.WithRequireLeader(t.Context()))
		require.NoError(t, err)

		_, err = keepAliveClient.Recv()
		require.Equal(t, rpctypes.ErrNoLeader.Error(), rpctypes.ErrorDesc(err))
		// Skip metric check in proxy mode - metrics are recorded on the proxy, not the etcd server.
		if !integration.ThroughProxy {
			require.Equal(t, prevUnavailableCount+1, getLeaseKeepAliveMetric(t, follower, "Unavailable"))
		}
	})

	// Client receives NoLeader error after the waitLeader() timed out in LeaseRenew().
	t.Run("catches NoLeader error without WithRequireLeader", func(t *testing.T) {
		leader, follower, anotherFollower := setupLeaseForwardingCluster(t)
		leaderClient := integration.ToGRPC(leader.Client).Lease
		followerClient := integration.ToGRPC(follower.Client).Lease