func (dht *IpfsDHT) handleAddProvider(p peer.Peer, pmes *pb.Message) (*pb.Message, error) {
	key := u.Key(pmes.GetKey())

	log.Debugf("%s adding %s as a provider for '%s'\n", dht.self, p, peer.ID(key))

	// add provider should use the address given in the message
	for _, pb := range pmes.GetProviderPeers() {
		pid := peer.ID(pb.GetId())
		if pid.Equal(p.ID()) {

			addr, err := pb.Address()
			if err != nil {
				log.Errorf("provider %s error with address %s", p, *pb.Addr)
				continue
			}

			log.Infof("received provider %s %s for %s", p, addr, key)
			p.AddAddress(addr)
			dht.providers.AddProvider(key, p)

		} else {
			log.Errorf("handleAddProvider received provider %s from %s", pid, p)
		}
	}

	return pmes, nil // send back same msg as confirmation.
}

// Handshake3Update updates local knowledge with the information in the
// handshake3 msg we received from remote client.
func Handshake3Update(lpeer, rpeer peer.Peer, msg *pb.Handshake3) (*Handshake3Result, error) {
	res := &Handshake3Result{}

	// our observed address
	observedAddr, err := ma.NewMultiaddrBytes(msg.GetObservedAddr())
	if err != nil {
		return res, err
	}
	if lpeer.AddAddress(observedAddr) {
		log.Infof("(nat) added new local, remote-observed address: %s", observedAddr)
	}
	res.LocalObservedAddress = observedAddr

	// remote's reported addresses
	for _, a := range msg.GetListenAddrs() {
		addr, err := ma.NewMultiaddrBytes(a)
		if err != nil {
			err = fmt.Errorf("remote peer address not a multiaddr: %s", err)
			log.Errorf("Handshake3 error %s", err)
			return res, err
		}
		rpeer.AddAddress(addr)
		res.RemoteListenAddresses = append(res.RemoteListenAddresses, addr)
	}

	return res, nil
}

func printPeer(p peer.Peer) (interface{}, error) {
	if p == nil {
		return nil, errors.New("Attempted to print nil peer!")
	}
	info := new(IdOutput)

	info.ID = p.ID().String()
	if p.PubKey() == nil {
		return nil, errors.New(`peer publickey not populated on offline runs,
please run the daemon to use ipfs id!`)
	}
	pkb, err := p.PubKey().Bytes()
	if err != nil {
		return nil, err
	}
	info.PublicKey = base64.StdEncoding.EncodeToString(pkb)
	for _, a := range p.Addresses() {
		info.Addresses = append(info.Addresses, a.String())
	}

	agent, protocol := p.GetVersions()
	info.AgentVersion = agent
	info.ProtocolVersion = protocol

	return info, nil
}

// peerMultiConn returns the MultiConn responsible for handling this peer.
// if there is none, it creates one and returns it. Note that timeouts
// and connection teardowns will remove it.
func (s *Swarm) peerMultiConn(p peer.Peer) (*conn.MultiConn, error) {

	s.connsLock.Lock()
	mc, found := s.conns[p.Key()]
	if found {
		s.connsLock.Unlock()
		return mc, nil
	}

	// multiconn doesn't exist, make a new one.
	mc, err := conn.NewMultiConn(s.Context(), s.local, p, nil)
	if err != nil {
		s.connsLock.Unlock()
		log.Errorf("error creating multiconn: %s", err)
		return nil, err
	}
	s.conns[p.Key()] = mc
	s.connsLock.Unlock()

	// kick off reader goroutine
	s.Children().Add(1)
	mc.Children().Add(1) // child of Conn as well.
	go s.fanInSingle(mc)
	return mc, nil
}

func (dht *IpfsDHT) handleFindPeer(p peer.Peer, pmes *pb.Message) (*pb.Message, error) {
	resp := pb.NewMessage(pmes.GetType(), "", pmes.GetClusterLevel())
	var closest []peer.Peer

	// if looking for self... special case where we send it on CloserPeers.
	if peer.ID(pmes.GetKey()).Equal(dht.self.ID()) {
		closest = []peer.Peer{dht.self}
	} else {
		closest = dht.betterPeersToQuery(pmes, CloserPeerCount)
	}

	if closest == nil {
		log.Errorf("handleFindPeer: could not find anything.")
		return resp, nil
	}

	var withAddresses []peer.Peer
	for _, p := range closest {
		if len(p.Addresses()) > 0 {
			withAddresses = append(withAddresses, p)
		}
	}

	for _, p := range withAddresses {
		log.Debugf("handleFindPeer: sending back '%s'", p)
	}
	resp.CloserPeers = pb.PeersToPBPeers(withAddresses)
	return resp, nil
}

// Dial connects to a particular peer, over a given network
// Example: d.Dial(ctx, "udp", peer)
func (d *Dialer) Dial(ctx context.Context, network string, remote peer.Peer) (Conn, error) {
	raddr := remote.NetAddress(network)
	if raddr == nil {
		return nil, debugerror.Errorf("No remote address for network %s", network)
	}
	return d.DialAddr(ctx, raddr, remote)
}

func (n *network) Adapter(p peer.Peer) bsnet.BitSwapNetwork {
	client := &networkClient{
		local:   p,
		network: n,
	}
	n.clients[p.Key()] = client
	return client
}

// ledger lazily instantiates a ledger
func (s *strategist) ledger(p peer.Peer) *ledger {
	l, ok := s.ledgerMap[peerKey(p.Key())]
	if !ok {
		l = newLedger(p, s.strategyFunc)
		s.ledgerMap[peerKey(p.Key())] = l
	}
	return l
}

func peerIsPartner(p peer.Peer, s Strategy) bool {
	for _, partner := range s.Peers() {
		if partner.Key() == p.Key() {
			return true
		}
	}
	return false
}

func (ps *peerSet) AddIfSmallerThan(p peer.Peer, maxsize int) bool {
	var success bool
	ps.lk.Lock()
	if _, ok := ps.ps[string(p.ID())]; !ok && len(ps.ps) < maxsize {
		success = true
		ps.ps[string(p.ID())] = true
	}
	ps.lk.Unlock()
	return success
}

func (pq *distancePQ) Enqueue(p peer.Peer) {
	pq.Lock()
	defer pq.Unlock()

	distance := ks.XORKeySpace.Key(p.ID()).Distance(pq.from)

	heap.Push(&pq.heap, &peerMetric{
		peer:   p,
		metric: distance,
	})
}

func (rs *hashTable) Announce(p peer.Peer, k u.Key) error {
	rs.lock.Lock()
	defer rs.lock.Unlock()

	_, ok := rs.providers[k]
	if !ok {
		rs.providers[k] = make(peer.Map)
	}
	rs.providers[k][p.Key()] = p
	return nil
}

// TODO(brian): handle errors
func (bs *bitswap) ReceiveMessage(ctx context.Context, p peer.Peer, incoming bsmsg.BitSwapMessage) (
	peer.Peer, bsmsg.BitSwapMessage) {
	log.Debugf("ReceiveMessage from %v", p.Key())
	log.Debugf("Message wantlist: %v", incoming.Wantlist())

	if p == nil {
		log.Error("Received message from nil peer!")
		// TODO propagate the error upward
		return nil, nil
	}
	if incoming == nil {
		log.Error("Got nil bitswap message!")
		// TODO propagate the error upward
		return nil, nil
	}

	// Record message bytes in ledger
	// TODO: this is bad, and could be easily abused.
	// Should only track *useful* messages in ledger
	bs.strategy.MessageReceived(p, incoming) // FIRST

	for _, block := range incoming.Blocks() {
		// TODO verify blocks?
		if err := bs.blockstore.Put(&block); err != nil {
			continue // FIXME(brian): err ignored
		}
		bs.notifications.Publish(block)
		err := bs.HasBlock(ctx, block)
		if err != nil {
			log.Warningf("HasBlock errored: %s", err)
		}
	}

	message := bsmsg.New()
	for _, wanted := range bs.wantlist.Keys() {
		message.AddWanted(wanted)
	}
	for _, key := range incoming.Wantlist() {
		// TODO: might be better to check if we have the block before checking
		//			if we should send it to someone
		if bs.strategy.ShouldSendBlockToPeer(key, p) {
			if block, errBlockNotFound := bs.blockstore.Get(key); errBlockNotFound != nil {
				continue
			} else {
				message.AddBlock(*block)
			}
		}
	}
	defer bs.strategy.MessageSent(p, message)

	log.Debug("Returning message.")
	return p, message
}

// CloseConnection removes a given peer from swarm + closes the connection
func (s *Swarm) CloseConnection(p peer.Peer) error {
	c := s.GetConnection(p.ID())
	if c == nil {
		return u.ErrNotFound
	}

	s.connsLock.Lock()
	delete(s.conns, u.Key(p.ID()))
	s.connsLock.Unlock()

	return c.Close()
}

func peerToPBPeer(p peer.Peer) *Message_Peer {
	pbp := new(Message_Peer)
	addrs := p.Addresses()
	if len(addrs) == 0 || addrs[0] == nil {
		pbp.Addr = proto.String("")
	} else {
		addr := addrs[0].String()
		pbp.Addr = &addr
	}
	pid := string(p.ID())
	pbp.Id = &pid
	return pbp
}

func (r *dhtQueryRunner) addPeerToQuery(next peer.Peer, benchmark peer.Peer) {
	if next == nil {
		// wtf why are peers nil?!?
		log.Error("Query getting nil peers!!!\n")
		return
	}

	// if new peer further away than whom we got it from, bother (loops)
	if benchmark != nil && kb.Closer(benchmark.ID(), next.ID(), r.query.key) {
		return
	}

	// if already seen, no need.
	r.Lock()
	_, found := r.peersSeen[next.Key()]
	if found {
		r.Unlock()
		return
	}
	r.peersSeen[next.Key()] = next
	r.Unlock()

	log.Debugf("adding peer to query: %v\n", next)

	// do this after unlocking to prevent possible deadlocks.
	r.peersRemaining.Increment(1)
	select {
	case r.peersToQuery.EnqChan <- next:
	case <-r.ctx.Done():
	}
}

// DialAddr connects to a peer over a particular address
// Ensures raddr is part of peer.Addresses()
// Example: d.DialAddr(ctx, peer.Addresses()[0], peer)
func (d *Dialer) DialAddr(ctx context.Context, raddr ma.Multiaddr, remote peer.Peer) (Conn, error) {

	found := false
	for _, addr := range remote.Addresses() {
		if addr.Equal(raddr) {
			found = true
		}
	}
	if !found {
		return nil, debugerror.Errorf("address %s is not in peer %s", raddr, remote)
	}

	network, _, err := manet.DialArgs(raddr)
	if err != nil {
		return nil, err
	}

	laddr := d.LocalPeer.NetAddress(network)
	if laddr == nil {
		return nil, debugerror.Errorf("No local address for network %s", network)
	}

	if strings.HasPrefix(raddr.String(), "/ip4/0.0.0.0") {
		return nil, debugerror.Errorf("Attempted to connect to zero address: %s", raddr)
	}

	remote.SetType(peer.Remote)
	remote, err = d.Peerstore.Add(remote)
	if err != nil {
		log.Errorf("Error putting peer into peerstore: %s", remote)
	}

	// TODO: try to get reusing addr/ports to work.
	// madialer := manet.Dialer{LocalAddr: laddr}
	madialer := manet.Dialer{}

	log.Infof("%s dialing %s %s", d.LocalPeer, remote, raddr)
	maconn, err := madialer.Dial(raddr)
	if err != nil {
		return nil, err
	}

	c, err := newSingleConn(ctx, d.LocalPeer, remote, maconn)
	if err != nil {
		return nil, err
	}

	return newSecureConn(ctx, c, d.Peerstore)
}

// Dial connects to a peer.
//
// The idea is that the client of Swarm does not need to know what network
// the connection will happen over. Swarm can use whichever it choses.
// This allows us to use various transport protocols, do NAT traversal/relay,
// etc. to achive connection.
//
// For now, Dial uses only TCP. This will be extended.
func (s *Swarm) Dial(peer peer.Peer) (conn.Conn, error) {
	if peer.ID().Equal(s.local.ID()) {
		return nil, errors.New("Attempted connection to self!")
	}

	// check if we already have an open connection first
	c := s.GetConnection(peer.ID())
	if c != nil {
		return c, nil
	}

	// check if we don't have the peer in Peerstore
	peer, err := s.peers.Add(peer)
	if err != nil {
		return nil, err
	}

	// open connection to peer
	d := &conn.Dialer{
		LocalPeer: s.local,
		Peerstore: s.peers,
	}

	// try to connect to one of the peer's known addresses.
	// for simplicity, we do this sequentially.
	// A future commit will do this asynchronously.
	for _, addr := range peer.Addresses() {
		c, err = d.DialAddr(s.Context(), addr, peer)
		if err == nil {
			break
		}
	}
	if err != nil {
		return nil, err
	}

	c, err = s.connSetup(c)
	if err != nil {
		c.Close()
		return nil, err
	}

	// TODO replace the TODO ctx with a context passed in from caller
	log.Event(context.TODO(), "dial", peer)
	return c, nil
}

func setupDHT(ctx context.Context, t *testing.T, p peer.Peer) *IpfsDHT {
	peerstore := peer.NewPeerstore()

	dhts := netservice.NewService(ctx, nil) // nil handler for now, need to patch it
	net, err := inet.NewIpfsNetwork(ctx, p.Addresses(), p, peerstore, &mux.ProtocolMap{
		mux.ProtocolID_Routing: dhts,
	})
	if err != nil {
		t.Fatal(err)
	}

	d := NewDHT(ctx, p, peerstore, net, dhts, ds.NewMapDatastore())
	dhts.SetHandler(d)
	d.Validators["v"] = func(u.Key, []byte) error {
		return nil
	}
	return d
}

// TODO should this be completely asynchronous?
// TODO what does the network layer do with errors received from services?
func (n *network) SendMessage(
	ctx context.Context,
	from peer.Peer,
	to peer.Peer,
	message bsmsg.BitSwapMessage) error {

	receiver, ok := n.clients[to.Key()]
	if !ok {
		return errors.New("Cannot locate peer on network")
	}

	// nb: terminate the context since the context wouldn't actually be passed
	// over the network in a real scenario

	go n.deliver(receiver, from, message)

	return nil
}