CtPtr ProtocolV2::handle_client_ident(ceph::bufferlist &payload)
{
  ldout(cct, 20) << __func__
		 << " payload.length()=" << payload.length() << dendl;

  if (state != SESSION_ACCEPTING) {
    lderr(cct) << __func__ << " not in session accept state!" << dendl;
    return _fault();
  }

  auto client_ident = ClientIdentFrame::Decode(payload);

  ldout(cct, 5) << __func__ << " received client identification:"
                << " addrs=" << client_ident.addrs()
		            << " target=" << client_ident.target_addr()
                << " gid=" << client_ident.gid()
                << " global_seq=" << client_ident.global_seq()
                << " features_supported=" << std::hex
                << client_ident.supported_features()
                << " features_required=" << client_ident.required_features()
                << " flags=" << client_ident.flags()
                << " cookie=" << client_ident.cookie() << std::dec << dendl;

  if (client_ident.addrs().empty() ||
      client_ident.addrs().front() == entity_addr_t()) {
    ldout(cct,5) << __func__ << " oops, client_ident.addrs() is empty" << dendl;
    return _fault();  // a v2 peer should never do this
  }
  if (!messenger->get_myaddrs().contains(client_ident.target_addr())) {
    ldout(cct,5) << __func__ << " peer is trying to reach "
		 << client_ident.target_addr()
		 << " which is not us (" << messenger->get_myaddrs() << ")"
		 << dendl;
    return _fault();
  }

  connection->set_peer_addrs(client_ident.addrs());
  connection->target_addr = connection->_infer_target_addr(client_ident.addrs());

  peer_name = entity_name_t(connection->get_peer_type(), client_ident.gid());
  connection->set_peer_id(client_ident.gid());

  client_cookie = client_ident.cookie();

  uint64_t feat_missing =
    (connection->policy.features_required | msgr2_required) &
    ~(uint64_t)client_ident.supported_features();
  if (feat_missing) {
    ldout(cct, 1) << __func__ << " peer missing required features " << std::hex
                  << feat_missing << std::dec << dendl;
    auto ident_missing_features =
        IdentMissingFeaturesFrame::Encode(feat_missing);

    return WRITE(ident_missing_features, "ident missing features", read_frame);
  }

  connection_features =
      client_ident.supported_features() & connection->policy.features_supported;

  peer_global_seq = client_ident.global_seq();

  if (connection->policy.server &&
      connection->policy.lossy &&
      !connection->policy.register_lossy_clients) {
    // incoming lossy client, no need to register this connection
  } else {
    // Looks good so far, let's check if there is already an existing connection
    // to this peer.
    connection->lock.unlock();
    AsyncConnectionRef existing = messenger->lookup_conn(
      *connection->peer_addrs);

    if (existing &&
	existing->protocol->proto_type != 2) {
      ldout(cct,1) << __func__ << " existing " << existing << " proto "
		   << existing->protocol.get() << " version is "
		   << existing->protocol->proto_type << ", marking down"
		   << dendl;
      existing->mark_down();
      existing = nullptr;
    }

    connection->inject_delay();

    connection->lock.lock();
    if (state != SESSION_ACCEPTING) {
      ldout(cct, 1) << __func__
		    << " state changed while accept, it must be mark_down"
		    << dendl;
      ceph_assert(state == CLOSED);
      return _fault();
    }

    if (existing) {
      return handle_existing_connection(existing);
    }
  }

  // if everything is OK reply with server identification
  return send_server_ident();
}