cjdns/dht/dhtcore/NodeStore.c

/* vim: set expandtab ts=4 sw=4: */
/*
 * You may redistribute this program and/or modify it under the terms of
 * the GNU General Public License as published by the Free Software Foundation,
 * either version 3 of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
#include "crypto/AddressCalc.h"
#include "dht/Address.h"
#include "dht/dhtcore/Node.h"
#include "dht/dhtcore/NodeStore.h"
#include "dht/dhtcore/NodeList.h"
#include "util/AddrTools.h"
#include "util/Assert.h"
#include "util/Bits.h"
#include "util/log/Log.h"
#include "util/version/Version.h"
#include "switch/NumberCompress.h"
#include "switch/LabelSplicer.h"
#include "util/Gcc.h"

#include <stdbool.h>
#include <inttypes.h>
#include <tree.h>

/** A list of DHT nodes. */
struct NodeStore_pvt
{
    struct NodeStore pub;

    /** A fake link where we are both the parent and child. */
    struct Node_Link* selfLink;

    /** A tree containing all nodes ordered by ipv6 */
    struct NodeRBTree {
        struct Node_Two* rbh_root;
    } nodeTree;

    struct Allocator* alloc;

    /** The maximum number of nodes which can be allocated. TODO: make use of */
    int capacity;

    /**
     * The links to be freed next time freePendingLinks() is called.
     */
    struct Node_Link* linksToFree;

    /** The means for this node store to log. */
    struct Log* logger;

    Identity
};

// My memory is really bad
#define A_COMES_FIRST 1
#define B_COMES_FIRST -1
static int comparePeers(const struct Node_Link* la, const struct Node_Link* lb)
{
    Identity_check(lb);
    uint64_t a = la->cannonicalLabel;
    uint64_t b = lb->cannonicalLabel;

    int log2Diff = Bits_log2x64(b) - Bits_log2x64(a);
    if (log2Diff) {
        return log2Diff;
    }
    if (Bits_bitReverse64(a) < Bits_bitReverse64(b)) {
        return A_COMES_FIRST;
    } else if (a == b) {
        return 0;
    }
    return B_COMES_FIRST;
}

RB_GENERATE_STATIC(PeerRBTree, Node_Link, peerTree, comparePeers)

static int compareNodes(const struct Node_Two* na, const struct Node_Two* nb)
{
    Identity_check(nb);
    int ret;
    ret = Address_xorcmp(0, na->address.ip6.ints.one_be, nb->address.ip6.ints.one_be);
    if (ret) { return ret; }
    ret = Address_xorcmp(0, na->address.ip6.ints.two_be, nb->address.ip6.ints.two_be);
    if (ret) { return ret; }
    ret = Address_xorcmp(0, na->address.ip6.ints.three_be, nb->address.ip6.ints.three_be);
    if (ret) { return ret; }
    ret = Address_xorcmp(0, na->address.ip6.ints.four_be, nb->address.ip6.ints.four_be);
    return ret;
}

RB_GENERATE_STATIC(NodeRBTree, Node_Two, nodeTree, compareNodes)

static void freeLink(struct Node_Link* link, struct NodeStore_pvt* store)
{
    Allocator_realloc(store->alloc, link, 0);
    store->pub.linkCount--;
}

static struct Node_Link* getLink(struct NodeStore_pvt* store)
{
    store->pub.linkCount++;
    return Allocator_calloc(store->alloc, sizeof(struct Node_Link), 1);
}

static void logLink(struct NodeStore_pvt* store,
                    struct Node_Link* link,
                    char* message)
{
    #ifndef Log_DEBUG
        return;
    #endif
    uint8_t parent[40];
    uint8_t child[40];
    AddrTools_printIp(parent, link->parent->address.ip6.bytes);
    AddrTools_printIp(child, link->child->address.ip6.bytes);
    uint8_t path[20];
    AddrTools_printPath(path, link->cannonicalLabel);
    Log_debug(store->logger, "link[%s]->[%s] [%s] %s", parent, child, path, message);
}

static void _checkNode(struct Node_Two* node, struct NodeStore_pvt* store, char* file, int line)
{
    #ifndef PARANOIA
        return;
    #endif

    Assert_true(node->address.path ==
        EncodingScheme_convertLabel(store->pub.selfNode->encodingScheme,
                                    node->address.path,
                                    EncodingScheme_convertLabel_convertTo_CANNONICAL));

    struct Node_Link* link;
    for (link = node->reversePeers; link; link = link->nextPeer) {
        Assert_fileLine(link->child == node, file, line);
        Assert_fileLine(RB_FIND(PeerRBTree, &link->parent->peerTree, link) == link, file, line);
    }

    struct Node_Link* lastLink = NULL;
    RB_FOREACH_REVERSE(link, PeerRBTree, &node->peerTree) {
        Assert_fileLine(node->bestParent || link->child->bestParent != link, file, line);
        Assert_fileLine(link->parent == node, file, line);
        Assert_fileLine(link->child != node || link == store->selfLink, file, line);
        Assert_fileLine(!lastLink || link->cannonicalLabel != lastLink->cannonicalLabel,
                        file, line);
        Assert_fileLine(link->cannonicalLabel < UINT64_MAX && link->cannonicalLabel > 0,
                        file, line);
        struct Node_Link* rlink = NULL;
        for (rlink = link->child->reversePeers; rlink; rlink = rlink->nextPeer) {
            if (rlink == link) {
                break;
            }
        }
        Assert_fileLine(rlink && "child contains reverse link", file, line);
        lastLink = link;
    }

    if (node->bestParent) {
        Assert_fileLine(Node_getReach(node->bestParent->parent) > Node_getReach(node)
            || node == store->pub.selfNode, file, line);

        Assert_fileLine(node->address.path != UINT64_MAX, file, line);
        // Should never get as low as 512...
        Assert_fileLine(Node_getReach(node) > 512, file, line);

        struct Node_Two* nn = node;
        do {
            Assert_fileLine(
                LabelSplicer_routesThrough(nn->address.path, nn->bestParent->parent->address.path),
                file,
                line
            );
            nn = nn->bestParent->parent;
        } while (nn != store->pub.selfNode);

    } else {
        Assert_fileLine(node->address.path == UINT64_MAX, file, line);
        Assert_fileLine(Node_getReach(node) == 0, file, line);
    }
}
#define checkNode(node, store) _checkNode(node, store, Gcc_SHORT_FILE, Gcc_LINE)

static void _verifyNode(struct Node_Two* node, struct NodeStore_pvt* store, char* file, int line)
{
    #ifndef PARANOIA
        return;
    #endif
    // #1 check the node (do the basic checks)
    _checkNode(node, store, file, line);

    // #2 make sure all of the node's outgoing links are split properly
    struct Node_Link* link = NULL;
    RB_FOREACH_REVERSE(link, PeerRBTree, &node->peerTree) {
        // make sure any peers of this node are split properly
        struct Node_Link* linkB = link;
        struct Node_Link* linkC = link;
        RB_FOREACH_REVERSE_FROM(linkB, PeerRBTree, linkC) {
            if (linkB == link || link == store->selfLink) { continue; }
            Assert_fileLine(
                !LabelSplicer_routesThrough(linkB->cannonicalLabel, link->cannonicalLabel),
                file, line
            );
        }
    }

    // #3 make sure looking for the node by address will actually find the correct node.
    if (node->bestParent) {
        Assert_fileLine(node == NodeStore_closestNode(&store->pub, node->address.path), file, line);
    }
}
#define verifyNode(node, store) _verifyNode(node, store, Gcc_SHORT_FILE, Gcc_LINE)

// Verify is more thorough than check because it makes sure all links are split properly.
static void _verify(struct NodeStore_pvt* store, char* file, int line)
{
    #ifndef PARANOIA
        return;
    #endif
    Assert_true(store->pub.selfNode->bestParent == store->selfLink || !store->selfLink);
    struct Node_Two* nn = NULL;
    RB_FOREACH(nn, NodeRBTree, &store->nodeTree) {
        _verifyNode(nn, store, file, line);
    }
}
#define verify(store) _verify(store, Gcc_SHORT_FILE, Gcc_LINE)

static void _check(struct NodeStore_pvt* store, char* file, int line)
{
    #ifndef PARANOIA
        return;
    #endif
    Assert_true(store->pub.selfNode->bestParent == store->selfLink || !store->selfLink);
    struct Node_Two* nn = NULL;
    RB_FOREACH(nn, NodeRBTree, &store->nodeTree) {
        _checkNode(nn, store, file, line);
    }
}
#define check(store) _check(store, Gcc_SHORT_FILE, Gcc_LINE)

/**
 * Extend a route by splicing on another link.
 * This will modify the Encoding Form of the first Director in next section of the route to make
 * it's size greater than or equal to the size of the return route through the parent node in the
 * link.
 *
 * @param routeLabel the label for reaching the parent node
 * @param link the link to extend the route with
 * @param previousLinkEncoding the encoding used for the parent's interface back to the grandparent
 */
static uint64_t extendRoute(uint64_t routeLabel,
                            struct EncodingScheme* parentScheme,
                            uint64_t parentChildLabel,
                            int previousLinkEncoding)
{
    uint64_t next = parentChildLabel;
    int nextLinkEncoding = EncodingScheme_getFormNum(parentScheme, next);
    if (previousLinkEncoding > nextLinkEncoding) {
        next = EncodingScheme_convertLabel(parentScheme, next, previousLinkEncoding);
    }
    Assert_true(next != EncodingScheme_convertLabel_INVALID);
    return LabelSplicer_splice(next, routeLabel);
}

static void setReach(struct Node_Two* node, uint32_t newReach)
{
    if (newReach) {
        Assert_true(node->bestParent);
        Assert_true(node->address.path < UINT64_MAX);
        Assert_true(newReach > 512);
    } else {
        Assert_true(!node->bestParent);
        Assert_true(node->address.path == UINT64_MAX);
    }
    node->reach_ro = newReach;
}

static void unreachable(struct Node_Two* node, struct NodeStore_pvt* store)
{
    struct Node_Link* next = NULL;
    RB_FOREACH_REVERSE(next, PeerRBTree, &node->peerTree) {
        if (next->child->bestParent == next) { unreachable(next->child, store); }
    }
    node->bestParent = NULL;
    node->address.path = UINT64_MAX;
    setReach(node, 0);
}

/**
 * This is called when we have no idea what the reach should be for the next hop
 * because the path we previously used to get to it is broken and we need to use
 * a different one. Take a somewhat educated guess as to what it might be in a way
 * that will make the reach non-zero.
 */
static uint32_t guessReachOfChild(struct Node_Link* link)
{
    // return 3/4 of the parent's reach if it's 1 hop, 1/2 otherwise.
    uint32_t r = Node_getReach(link->parent) / 2;
    if (r < (1<<12)) {
        r = Node_getReach(link->parent) - 1;
    } else if (r < (1<<16)) {
        r = Node_getReach(link->parent) - Bits_log2x64(link->cannonicalLabel);
    }
    Assert_true(r < Node_getReach(link->parent) && r != 0);
    return r;
}

static int updateBestParentCycle(struct Node_Two* node,
                                 int cycle,
                                 int limit,
                                 uint32_t nextReach,
                                 struct NodeStore_pvt* store)
{
    Assert_always(cycle < 1000);
    if (cycle < limit) {
        int total = 0;
        struct Node_Link* next = NULL;
        RB_FOREACH_REVERSE(next, PeerRBTree, &node->peerTree) {
            if (next->child->bestParent == next && next->child != node) {
                total += updateBestParentCycle(next->child, cycle+1, limit, nextReach, store);
            }
        }
        return total;
    }

    struct Node_Link* newBestLink = node->bestParent;
    struct Node_Two* newBest = newBestLink->parent;

    uint64_t bestPath = extendRoute(newBest->address.path,
                                    newBest->encodingScheme,
                                    newBestLink->cannonicalLabel,
                                    newBest->bestParent->inverseLinkEncodingFormNumber);

    if (bestPath == UINT64_MAX) {
        unreachable(node, store);
        return 1;
    }

    /*#ifdef Log_DEBUG
        if (node->address.path != bestPath) {
            uint8_t pathStr[20];
            AddrTools_printPath(pathStr, bestPath);
            uint8_t addrStr[40];
            AddrTools_printIp(addrStr, node->address.ip6.bytes);
            Log_debug(store->logger, "New best path [%s@%s]", addrStr, pathStr);
        }
    #endif*/

    node->address.path = bestPath;
    if (!limit) {
        setReach(node, nextReach);
    }
    checkNode(node, store);
    return 1;
}

static void updateBestParent(struct Node_Two* node,
                             struct Node_Link* newBestParent,
                             uint32_t nextReach,
                             struct NodeStore_pvt* store)
{
    check(store);
    node->bestParent = newBestParent;
    Assert_true(newBestParent);
    //Assert_true(nextReach > 1023);
    Assert_true(nextReach < Node_getReach(newBestParent->parent));

    for (int i = 0; i < 10000; i++) {
        if (!updateBestParentCycle(node, 0, i, nextReach, store)) {
            check(store);
            return;
        }
    }
    Assert_true(0);
}

static void handleGoodNews(struct Node_Two* node,
                           uint32_t newReach,
                           struct NodeStore_pvt* store)
{
    // TODO: Paths longer than 1024 will blow up, handle more gracefully
    Assert_always(newReach != UINT32_MAX);
    Assert_always(newReach > 1023);

    Assert_true(newReach > Node_getReach(node));

    // The nodestore thinks it's unreachable, we can't very well update the reach.
    if (node->bestParent == NULL) { return; }

    if (newReach+1 > Node_getReach(node->bestParent->parent)) {
        handleGoodNews(node->bestParent->parent, newReach+1, store);
    }
    setReach(node, newReach);
    struct Node_Link* link = NULL;
    RB_FOREACH_REVERSE(link, PeerRBTree, &node->peerTree) {
        struct Node_Two* child = link->child;
        if (!child->bestParent || Node_getReach(child->bestParent->parent) < newReach) {
            uint32_t nextReach = guessReachOfChild(link);
            if (Node_getReach(child) > nextReach) { continue; }
            updateBestParent(child, link, nextReach, store);
        }
    }
}

/**
 * The news has hit (in handleBadNewsOne) and now all of the nodes in the affected zone have
 * been knocked down. Now lets see if there's a better path for any of them.
 */
static void handleBadNewsTwo(struct Node_Link* link, struct NodeStore_pvt* store)
{
    struct Node_Link* next = NULL;
    RB_FOREACH_REVERSE(next, PeerRBTree, &link->child->peerTree) {
        if (!next) { continue; }
        if (next->child->bestParent != next) { continue; }
        if (next == store->selfLink) { continue; }
        handleBadNewsTwo(next, store);
    }

    // node was relinked by a recursion of this function.
    if (link->child->bestParent != link) { return; }

    struct Node_Two* node = link->child;
    struct Node_Link* rp = link->child->reversePeers;
    struct Node_Link* best = node->bestParent;
    while (rp) {
        if (Node_getReach(rp->parent) >= Node_getReach(best->parent)) {
            if (Node_getReach(rp->parent) > Node_getReach(best->parent)
                || rp->parent->address.path < best->parent->address.path)
            {
                best = rp;
            }
        }
        rp = rp->nextPeer;
    }


    if (best == node->bestParent) { return; }

    uint32_t nextReach = guessReachOfChild(best);
    if (nextReach <= Node_getReach(node)) { return; }
    Assert_true(Node_getReach(node) < Node_getReach(best->parent));

    check(store);
    updateBestParent(node, best, nextReach, store);
    check(store);
}

/**
 * First thing we do is knock down everybody's reach.
 * This way they don't all cling to eachother for safety making
 * endless routing loops and stupid processing.
 */
static uint32_t handleBadNewsOne(struct Node_Link* link,
                                 uint32_t newReach,
                                 struct NodeStore_pvt* store)
{
    struct Node_Link* next = NULL;
    uint32_t highestRet = 0;
    RB_FOREACH_REVERSE(next, PeerRBTree, &link->child->peerTree) {
        if (next->child->bestParent != next) { continue; }
        if (next == store->selfLink) { continue; }
        if (Node_getReach(next->child) < newReach) { continue; }

        uint32_t ret = handleBadNewsOne(next, newReach, store);
        if (ret > highestRet) { highestRet = ret; }
    }
    if (!highestRet) { highestRet = newReach; }

    Assert_true(link->child != store->pub.selfNode);
    if (!highestRet) {
        unreachable(link->child, store);
    } else {
        setReach(link->child, highestRet);
    }

    if (highestRet < 1023) { highestRet = 1023; }
    return highestRet+1;
}

static void handleBadNews(struct Node_Two* node,
                          uint32_t newReach,
                          struct NodeStore_pvt* store)
{
    Assert_true(newReach < Node_getReach(node));

    // no bestParent implies a reach of 0
    Assert_true(node->bestParent);

    Assert_true(node->bestParent != store->selfLink);

    // might be destroyed by handleBadNewsOne()
    struct Node_Link* bp = node->bestParent;

    Assert_true(!newReach || newReach > 1023);
    handleBadNewsOne(node->bestParent, newReach, store);

    // If our bad news actually improved the reach number for the node (because it was previously
    // 0 and that node has children) then we need to handle it as good news as well.
    if (node->bestParent) {
        if (Node_getReach(node) >= Node_getReach(node->bestParent->parent)) {
            handleGoodNews(node->bestParent->parent, Node_getReach(node)+1, store);
        }
        Assert_true(Node_getReach(node) < Node_getReach(node->bestParent->parent));
    }

    check(store);

    handleBadNewsTwo(bp, store);

    check(store);
}

static void handleNews(struct Node_Two* node, uint32_t newReach, struct NodeStore_pvt* store)
{
    // This is because reach is used to prevent loops so it must be 1 more for each hop closer
    // to the root.
    if (newReach > (UINT32_MAX - 1024)) { newReach = (UINT32_MAX - 1024); }
    if (newReach < 1024) { newReach = 1024; }

    check(store);
    if (newReach < Node_getReach(node)) {
        handleBadNews(node, newReach, store);
        check(store);
    } else if (newReach > Node_getReach(node)) {
        handleGoodNews(node, newReach, store);
        check(store);
    }
}

static void unlinkNodes(struct Node_Link* link, struct NodeStore_pvt* store)
{
    struct Node_Two* child = Identity_check(link->child);
    struct Node_Two* parent = Identity_check(link->parent);
    check(store);

    // Change the best parent and path if necessary
    if (child->bestParent == link) {
        handleBadNews(child, 0, store);
    }

    if (child->bestParent == link) {
        unreachable(child, store);
    }

    check(store);

    // Remove the entry from the reversePeers
    struct Node_Link* current = child->reversePeers;
    struct Node_Link** currentP = &child->reversePeers;
    while (current) {
        if (current == link) {
            *currentP = current->nextPeer;
            break;
        }
        currentP = &(current->nextPeer);
        current = *currentP;
    }
    Assert_true(current);

    // Remove the RBTree entry
    Assert_ifParanoid(link == RB_FIND(PeerRBTree, &parent->peerTree, link));
    RB_REMOVE(PeerRBTree, &parent->peerTree, link);

    link->nextPeer = store->linksToFree;
    store->linksToFree = link;

    // prevent double-free of link.
    link->parent = NULL;
    link->child = NULL;

    check(store);
}

static void update(struct Node_Link* link,
                   int64_t linkStateDiff,
                   struct NodeStore_pvt* store)
{
    /** TODO: Link state is not taken into account yet
    if (linkStateDiff + link->linkState > UINT32_MAX) {
        link->linkState = UINT32_MAX;
        logLink(store, link, "link state set to maximum");
    } else if (linkStateDiff + link->linkState < 0) {
        link->linkState = UINT32_MAX;
        logLink(store, link, "link state set to zero");
    } else {
        link->linkState += linkStateDiff;
    }
    */
}

/**
 * Link two nodes in the graph together.
 * If a parent of the child node is also a parent of the parent node, they are
 * unlinked (the link is split and the child is inserted in the middle).
 *
 * @param parent the current end of the graph
 * @param child the new node to extend the graph
 * @param cannonicalLabel the label for getting from the parent to the child.
 * @param linkStateDiff how much to change the link state for this link.
 * @param store
 */
static struct Node_Link* linkNodes(struct Node_Two* parent,
                                   struct Node_Two* child,
                                   uint64_t cannonicalLabel,
                                   int64_t linkStateDiff,
                                   int inverseLinkEncodingFormNumber,
                                   uint64_t discoveredPath,
                                   struct NodeStore_pvt* store)
{
    check(store);

    #ifdef Log_DEBUG
        uint8_t parentIp[40];
        uint8_t childIp[40];
        AddrTools_printIp(parentIp, parent->address.ip6.bytes);
        AddrTools_printIp(childIp, child->address.ip6.bytes);
        uint8_t printedLabel[20];
        AddrTools_printPath(printedLabel, cannonicalLabel);
        Log_debug(store->logger, "Linking [%s] with [%s] with label fragment [%s]",
                  parentIp, childIp, printedLabel);
    #endif

    // It's ok to link a node with itself via some loopey route.
    // in practice it should never actually be used and it might yield some interesting
    // information when the link is split, self-routes are not allowed unless the self
    // link is being set up :)
    Assert_true(cannonicalLabel != 1 || store->selfLink == NULL);

    #ifdef PARANOIA
        uint64_t definitelyCannonical =
            EncodingScheme_convertLabel(parent->encodingScheme,
                                        cannonicalLabel,
                                        EncodingScheme_convertLabel_convertTo_CANNONICAL);
        Assert_true(definitelyCannonical == cannonicalLabel);
    #endif

    struct Node_Link* link;
    RB_FOREACH_REVERSE(link, PeerRBTree, &parent->peerTree) {
        Identity_check(link);
        if (link->child == child) {
            if (link->cannonicalLabel != cannonicalLabel) {
                // multiple paths between A and B are ok because they
                // will have divergent paths following the first director.
                continue;

            } else if (link->inverseLinkEncodingFormNumber != inverseLinkEncodingFormNumber) {
                logLink(store, link, "Relinking nodes with different encoding form");
                // This can happen when C renumbers but B->C is the same because B did
                // not renumber, EG: if C restarts.
                link->inverseLinkEncodingFormNumber = inverseLinkEncodingFormNumber;
            }
            update(link, linkStateDiff, store);
            return link;
        }
    }

    struct Node_Link dummy = { .cannonicalLabel = cannonicalLabel };
    link = Identity_ncheck(RB_FIND(PeerRBTree, &parent->peerTree, &dummy));
    if (link) {
        logLink(store, link, "Attempted to create alternate link with same label!");
        Assert_true(0);
        return link;
    }

    link = getLink(store);

    // set it up
    link->cannonicalLabel = cannonicalLabel;
    link->inverseLinkEncodingFormNumber = inverseLinkEncodingFormNumber;
    link->child = child;
    link->parent = parent;
    link->discoveredPath = discoveredPath;
    Identity_set(link);

    // reverse link
    link->nextPeer = child->reversePeers;
    child->reversePeers = link;

    // forward link
    Assert_ifParanoid(!RB_FIND(PeerRBTree, &parent->peerTree, link));
    RB_INSERT(PeerRBTree, &parent->peerTree, link);

    if (!child->bestParent) {
        if (parent->bestParent) {
            updateBestParent(child, link, guessReachOfChild(link), store);
        } else {
            unreachable(child, store);
        }
    }

    // update the child's link state and possibly change it's preferred path
    update(link, linkStateDiff, store);

    check(store);
    return link;
}

/**
 * Find the closest node to the given path.
 * Pay especially close attention to the comments in this function, they're critical to
 * understanting what it actually does.
 *
 * @param path the path to the node which we want the closest node to.
 * @param output a pointer to be set to the link to the closest node.
 * @param hops a pointer to an integer which is initially the limit on the number of allowed hops.
 *             If there are more than this number of hope in the label, the search will terminate
 *             early. At the end this will be set to the actual number of hops until the find.
 * @param store
 * @return the label fragment linking outputNode with the given path.
 */
#define findClosest_INVALID (~((uint64_t)0))
static uint64_t findClosest(const uint64_t path,
                            struct Node_Link** output,
                            uint32_t* hops,
                            struct Node_Link* parentLink,
                            struct NodeStore_pvt* store)
{
    struct Node_Link tmpl = {
        // The path from us is always cannonical
        .cannonicalLabel = path
    };

    struct Node_Link* nextLink;
    struct Node_Link* link = parentLink;
    uint32_t actualHops = 0;
    for (; !hops || actualHops < *hops; actualHops++) {

        // Then we cannoicalize the child's Director
        if (link != parentLink) {

            // First we splice off the parent's Director leaving the child's Director.
            tmpl.cannonicalLabel =
                LabelSplicer_unsplice(tmpl.cannonicalLabel, link->cannonicalLabel);

            int formNum =
                EncodingScheme_getFormNum(link->child->encodingScheme, tmpl.cannonicalLabel);
            // Check that they didn't send us an obviously invalid route.
            if (formNum < link->inverseLinkEncodingFormNumber) {
                Assert_ifTesting(!"invalid route");
                Log_info(store->logger, "Invalid route");
                return findClosest_INVALID;
            }

            uint64_t cannonical =
                EncodingScheme_convertLabel(link->child->encodingScheme,
                                            tmpl.cannonicalLabel,
                                            EncodingScheme_convertLabel_convertTo_CANNONICAL);

            // Check that they didn't waste space by sending an oversize encoding form.
            if (formNum > link->inverseLinkEncodingFormNumber
                && cannonical != tmpl.cannonicalLabel)
            {
                Assert_ifTesting(!"wasting space");
                Log_info(store->logger, "Wasted space");
                return findClosest_INVALID;
            }
            tmpl.cannonicalLabel = cannonical;
        }

        Assert_true(tmpl.cannonicalLabel != EncodingScheme_convertLabel_INVALID);

        // Then we search for the next peer in the path
        nextLink = Identity_ncheck(RB_NFIND(PeerRBTree, &link->child->peerTree, &tmpl));

        while (nextLink
            && !LabelSplicer_routesThrough(tmpl.cannonicalLabel, nextLink->cannonicalLabel))
        {
            //logLink(store, nextLink, "GETTING NEXT LINK");
            nextLink = Identity_ncheck(RB_NEXT(PeerRBTree, NULL, nextLink));
        }

        if (!nextLink || nextLink == store->selfLink) {
            // ignore the comments, they're mostly wrong anyway
            break;
        }

        Identity_check(nextLink);
        Assert_true(nextLink->child->encodingScheme);

        if (tmpl.cannonicalLabel == nextLink->cannonicalLabel) {
            //logLink(store, nextLink, "Exact match");
            tmpl.cannonicalLabel = 1;
            *output = nextLink;
            if (hops) { *hops = actualHops; }
            return 1;
        }

        if (!LabelSplicer_routesThrough(tmpl.cannonicalLabel, nextLink->cannonicalLabel)) {
            // child of next link is not in the path, we reached the end.
            break;
        }

        /*#ifdef Log_DEBUG
            uint8_t labelA[20];
            uint8_t labelB[20];
            uint8_t searchingFor[20];
            AddrTools_printPath(labelA, tmpl.cannonicalLabel);
            AddrTools_printPath(searchingFor, path);
            AddrTools_printPath(labelB, link->cannonicalLabel);
            Log_debug(store->logger, "[%s] is behind [%s] searching for [%s]",
                      labelA, labelB, searchingFor);
        #endif*/

        link = nextLink;
    }

    /*#ifdef Log_DEBUG
        uint8_t labelA[20];
        uint8_t labelB[20] = "NONE";
        uint8_t labelC[20];
        AddrTools_printPath(labelA, tmpl.cannonicalLabel);
        if (nextLink) {
            AddrTools_printPath(labelB, nextLink->cannonicalLabel);
        }
        AddrTools_printPath(labelC, link->cannonicalLabel);
        Log_debug(store->logger, "[%s] is not behind [%s] closest: [%s]", labelA, labelB, labelC);
    #endif*/

    Assert_true(tmpl.cannonicalLabel);/// TODO remove this
    *output = link;
    if (hops) { *hops = actualHops; }
    return tmpl.cannonicalLabel;
}

static struct Node_Two* nodeForIp(struct NodeStore_pvt* store, uint8_t ip[16])
{
    struct Node_Two fakeNode;
    Identity_set(&fakeNode);
    Bits_memcpyConst(fakeNode.address.ip6.bytes, ip, 16);
    return Identity_ncheck(RB_FIND(NodeRBTree, &store->nodeTree, &fakeNode));
}

static bool isAncestorOf(struct NodeStore_pvt* store,
                         struct Node_Two* maybeAncestor,
                         struct Node_Two* maybeDecendent)
{
    struct Node_Link* parent = maybeDecendent->bestParent;
    for (int i = 0; i < 1000; i++) {
        if (!maybeDecendent->bestParent) { return false; }
        if (store->pub.selfNode == parent->parent) { return false; }
        if (maybeAncestor == parent->parent) { return true; }
        parent = parent->parent->bestParent;
    }
    Assert_always(0);
}

static void freePendingLinks(struct NodeStore_pvt* store)
{
    struct Node_Link* link;
    while ((link = store->linksToFree)) {
        store->linksToFree = link->nextPeer;
        freeLink(link, store);
    }
}

static struct Node_Link* discoverLink(struct NodeStore_pvt* store,
                                      struct Node_Link* closestKnown,
                                      uint64_t pathKnownParentChild,
                                      struct Node_Two* child,
                                      uint64_t discoveredPath,
                                      int inverseLinkEncodingFormNumber)
{
    // Make sure this link cannot be split before beginning.
    struct Node_Link* closest = NULL;
    uint64_t pathParentChild =
        findClosest(pathKnownParentChild, &closest, NULL, closestKnown, store);

    if (pathParentChild == findClosest_INVALID) {
        return NULL;
    }

    struct Node_Two* parent = closest->child;

    #ifdef Log_DEBUG
    {
        uint8_t parentStr[40];
        uint8_t childStr[40];
        uint8_t pathStr[20];

        AddrTools_printIp(parentStr, parent->address.ip6.bytes);
        AddrTools_printIp(childStr, child->address.ip6.bytes);
        AddrTools_printPath(pathStr, pathParentChild);
        Log_debug(store->logger, "discoverLink( [%s]->[%s] [%s] )", parentStr, childStr, pathStr);
    }
    #endif

    if (parent == child) {
        if (pathParentChild == 1) {
            // Link is already known.
            update(closest, 0, store);
            //Log_debug(store->logger, "Already known");
            return closest;
        }

        Log_debug(store->logger, "Loopey route");
        // lets not bother storing this link, a link with the same parent and child is
        // invalid according to verify() and it's just going to take up space in the store
        // we'll return closest which is a perfectly valid path to the same node.
        return closest;
    }

    while (pathParentChild == 1) {
        logLink(store, closest, "Node at end of path appears to have changed");

        // This should never happen for a direct peer or for a direct decendent in
        // a split link.
        Assert_always(closestKnown != closest);

        unlinkNodes(closest, store);
        pathParentChild =
            findClosest(pathKnownParentChild, &closest, NULL, closestKnown, store);

        if (pathParentChild != findClosest_INVALID) {
            // TODO: handle this in some way other than just failing to install the route.
            logLink(store, closestKnown, "Apparently the reverse link encoding scheme for "
                                         "link has changed.");
            return NULL;
        }

        parent = closest->child;
        check(store);
    }

    // link parent to child
    // TODO: linking every node with 0 link state, this can't be right.
    struct Node_Link* parentLink = linkNodes(parent,
                                             child,
                                             pathParentChild,
                                             0,
                                             inverseLinkEncodingFormNumber,
                                             discoveredPath,
                                             store);

    if (!RB_FIND(NodeRBTree, &store->nodeTree, child)) {
        checkNode(child, store);
        RB_INSERT(NodeRBTree, &store->nodeTree, child);
        store->pub.nodeCount++;
    }

    check(store);

    // Check whether the parent is already linked with a node which is "behind" the child.
    // previous appears to be a "sibling link" to the closest->node link but in reality the
    // previous link should be split and node should be inserted in the middle.
    struct Node_Link* splitLink = RB_MIN(PeerRBTree, &parent->peerTree);
    while (splitLink) {
        if (splitLink->cannonicalLabel <= pathParentChild) {
            if (splitLink->cannonicalLabel == pathParentChild) {
                Assert_true(splitLink->child == child);
                splitLink = PeerRBTree_RB_NEXT(splitLink);
                continue;
            } else {
                // Since they're in order, definitely not found.
                break;
            }
        }

        if (!LabelSplicer_routesThrough(splitLink->cannonicalLabel, pathParentChild)) {
            splitLink = PeerRBTree_RB_NEXT(splitLink);
            continue;
        }

        struct Node_Two* grandChild = splitLink->child;
        // unsplice and cannonicalize so we now have a path from child to grandchild
        uint64_t childToGrandchild =
            LabelSplicer_unsplice(splitLink->cannonicalLabel, pathParentChild);
        childToGrandchild =
            EncodingScheme_convertLabel(child->encodingScheme,
                                        childToGrandchild,
                                        EncodingScheme_convertLabel_convertTo_CANNONICAL);

        // just so we're on the same page here
        Assert_true(splitLink->parent == parent);
        Assert_true(childToGrandchild < UINT64_MAX);

        #ifdef Log_DEBUG
        {
            uint8_t parentStr[40];
            uint8_t childStr[40];
            uint8_t pathStr[20];

            AddrTools_printIp(parentStr, splitLink->parent->address.ip6.bytes);
            AddrTools_printIp(childStr, splitLink->child->address.ip6.bytes);
            AddrTools_printPath(pathStr, splitLink->cannonicalLabel);
            Log_debug(store->logger, "Splitting link [%s]->[%s] [%s]",
                      parentStr, childStr, pathStr);

            AddrTools_printIp(parentStr, splitLink->parent->address.ip6.bytes);
            AddrTools_printIp(childStr, child->address.ip6.bytes);
            AddrTools_printPath(pathStr, pathParentChild);
            Log_debug(store->logger, "New parent [%s]->[%s] [%s]", parentStr, childStr, pathStr);

            AddrTools_printIp(parentStr, child->address.ip6.bytes);
            AddrTools_printIp(childStr, splitLink->child->address.ip6.bytes);
            AddrTools_printPath(pathStr, childToGrandchild);
            Log_debug(store->logger, "New child [%s]->[%s] [%s]", parentStr, childStr, pathStr);
        }
        #endif

        if (child == grandChild) {
            // There's an existing link from the parent to the child and it loops
            // it takes a detour over to some other nodes and then comes back to the grandChild
            Log_debug(store->logger, "replace existing link which contains a loop...");
            goto done;
        }

        if (grandChild->bestParent == splitLink
            && Node_getReach(child) <= Node_getReach(grandChild))
        {
            // We know that the grandchild decends from the parent because splitLink is parent-->gc
            // Two possibilities:
            // someRoute-->child-->parent
            // someRoute-->parent-->child

            check(store);
            if (Node_getReach(parent) >= Node_getReach(child)) {
                // Parent definitely does not decend from child.
                Assert_true(Node_getReach(grandChild) < UINT32_MAX);
                updateBestParent(child, parentLink, Node_getReach(child), store);
            } else {
                // Child definitely does not decend from parent
                // Parent may decend from child, if it does we cannot safely re-root child
                // if not then we could but if we believe the reach of the child is better,
                // we might as well use the route which goes via the child rather than re-rooting
                // it anyway.
            }
            handleGoodNews(child, Node_getReach(grandChild)+1, store);

            // Node_getReach(parent) is by definition higher than Node_getReach(grandChild)
            // so if child is a decendent of grandChild then it should have been switched.
            Assert_true(Node_getReach(child) > Node_getReach(grandChild));

            check(store);
        }

        Assert_true(splitLink->cannonicalLabel != pathParentChild);
        Assert_true(childToGrandchild != 1);

        struct Node_Link* lcg = discoverLink(store,
                                             parentLink,
                                             childToGrandchild,
                                             grandChild,
                                             discoveredPath,
                                             splitLink->inverseLinkEncodingFormNumber);

        // so...
        // There is a chance... that in the recursion, the link we JUST CREATED (parentLink)
        // was split and unlinked. If that is so, we really should just return because
        // everything we were planning to do here has been done for us.
        if (!parentLink->child) {
            // return that last link along pathParentChild.
            struct Node_Link* link = NULL;
            findClosest(pathParentChild, &link, NULL, closest, store);
            Assert_always(link);
            return link;
        }

        if (grandChild->bestParent != splitLink) {
            // The link has been created and we don't care much about it because it's not
            // the best path.
            goto done;
        }

        if (!lcg) {
            // The path is probably broken... TODO: log or something
            goto done;
        }

        if (!lcg->parent->bestParent) {
            // The path is probably too long so it can't be represented, unreachable...
            goto done;
        }

        // Normally we would expect lcg->parent to be equal to child but because of the
        // findClosest() call at the beginning of the discoverLink() function, that is not
        // necessarily true. One or more nodes along the path childToGrandchild might
        // already be known, in which case lcg->parent will be the last known node along
        // that path.
        //
        // Worse, lcg->parent->bestParent might actually be grandChild or a decendent
        // thereof. Consider a path looking like this:
        // parent<-child<-Alice<-Bob Charlie->Dave->grandChild
        //
        // The best path to grandChild is obviously via child and therefor Charlie and Dave
        // are mistaken and this is a phantom loop.

        if (Node_getReach(lcg->parent) <= Node_getReach(grandChild)) {

            // I know, I repeat myself...
            Assert_true(Node_getReach(child) > Node_getReach(grandChild));

            if (isAncestorOf(store, grandChild, lcg->parent)) {

                // Again, just making dead sure that the child is not a decendent of the
                // grandChild, this should not be possible at this point
                Assert_true(parentLink->child != lcg->parent);

                // Now we're going to walk the path, when we encounter a node which decends from
                // grandChild, we're going to switch it to decend from the previous node along the
                // path.
                struct Node_Link* link = NULL;
                for (int limit = 1; ; limit++) {
                    int limitCpy = limit;
                    link = NULL;
                    findClosest(childToGrandchild,
                                &link,
                                &limitCpy,
                                parentLink,
                                store);
                    Assert_always(link);

                    if (link->child == grandChild) { break; }

                    // We should never get here because the if statement below should have
                    // handled this in the previous round.
                    Assert_true(Node_getReach(link->parent) >= Node_getReach(grandChild));

                    if (Node_getReach(link->child) <= Node_getReach(grandChild)) {
                        // Ok we're found a node whose pathQuality is less than grandChild's
                        // and this node is in the best path *to* the grandChild so it's quality
                        // needs to be increased and we need to swap it's bestParent so that we're
                        // sure it doesn't decend from grandChild.
                        if (link->child->bestParent != link) {
                            if (Node_getReach(link->parent) <= Node_getReach(grandChild)+1) {
                                handleGoodNews(link->parent, Node_getReach(grandChild)+2, store);
                            }
                            Assert_true(Node_getReach(link->parent) > Node_getReach(grandChild));
                            updateBestParent(link->child, link, Node_getReach(grandChild)+1, store);
                        }
                        if (Node_getReach(link->child) <= Node_getReach(grandChild)) {
                            handleGoodNews(link->child, Node_getReach(grandChild)+1, store);
                        }
                        Assert_true(Node_getReach(link->child) > Node_getReach(grandChild));
                    }
                }

                // Guess what!
                // grandChild *might* just show up MULTIPLE times in the path (phantom loop),
                // of so, link is now the first instance of grandChild and lcg is the final
                // instance (end of the path). We don't care much about the path beyond the first
                // instance of grandChild so we're just going to quietly swap the lcg for link
                // if it happens that they are not the same.
                lcg = link;
                Assert_true(Node_getReach(lcg->parent) > Node_getReach(grandChild));

            } else {
                handleGoodNews(lcg->parent, Node_getReach(grandChild)+1, store);
                Assert_true(Node_getReach(lcg->parent) > Node_getReach(grandChild));
            }

            Assert_true(Node_getReach(lcg->parent) > Node_getReach(grandChild));
            updateBestParent(grandChild, lcg, Node_getReach(grandChild), store);
        }

        // pfew
        done:

        check(store);

        // be careful!
        // splitLink might have been unlinked by the recursive discoverLink() call.
        if (splitLink->parent) {
            unlinkNodes(splitLink, store);
        }

        // link RB_NEXT might have also been freed by a recursive call to discoverLink()
        // so we'll just start over from the beginning and walk the list of links.
        splitLink = RB_MIN(PeerRBTree, &parent->peerTree);
    }

    check(store);
    return parentLink;
}

static struct Node_Two* whichIsWorse(struct Node_Two* one,
                                     struct Node_Two* two,
                                     struct NodeStore_pvt* store)
{
    if (one->address.protocolVersion != two->address.protocolVersion) {
        if (one->address.protocolVersion < Version_CURRENT_PROTOCOL) {
            if (two->address.protocolVersion >= Version_CURRENT_PROTOCOL) {
                return one;
            }
        } else if (two->address.protocolVersion < Version_CURRENT_PROTOCOL) {
            if (one->address.protocolVersion >= Version_CURRENT_PROTOCOL) {
                return two;
            }
        }
    }
    if (Address_closest(&store->pub.selfNode->address, &one->address, &two->address) > 0) {
        return one;
    }
    return two;
}

/**
 * We define the worst node as either node which is furthest in address space
 * from us which is either unreachable or, if no nodes are unreachable, has no
 * nodes for which it is the bestParent.
 * This metric will take into account both reach and address distance, although
 * reach is not directly fed into the function but rather the function relies on
 * the structure of the store which is affected by the reach.
 * O(2n)
 */
static struct Node_Two* getWorstNode(struct NodeStore_pvt* store)
{
    struct Node_Two* worst = NULL;
    struct Node_Two* nn = NULL;
    RB_FOREACH(nn, NodeRBTree, &store->nodeTree) {
        // first cycle we clear all markings as we go and set markings
        // so markings remain if they are behind us
        nn->marked = 0;
        if (nn->bestParent) {
            nn->bestParent->parent->marked = 1;
        } else if (!worst || whichIsWorse(nn, worst, store) == nn) {
            // this time around we're only addressing nodes which are unreachable.
            worst = nn;
        }
    }
    if (worst) { return worst; }
    RB_FOREACH(nn, NodeRBTree, &store->nodeTree) {
        // second cycle we set the markings as we go but if they are behind the
        // node which would have marked them, they are already set.
        if (nn->bestParent) {
            nn->bestParent->parent->marked = 1;
        }
        if (nn->marked) { continue; }
        if (!worst || whichIsWorse(nn, worst, store) == nn) {
            worst = nn;
        }
    }
    // somebody has to be at the end of the line, not *everyone* can be someone's best parent!
    Assert_true(worst);
    return worst;
}

static void destroyNode(struct Node_Two* node, struct NodeStore_pvt* store)
{
    struct Node_Link* link;
    RB_FOREACH(link, PeerRBTree, &node->peerTree) {
        Identity_check(link);
        unlinkNodes(link, store);
    }

    // optimization
    #ifndef PARANOIA
        node->bestParent = NULL;
        node->address.path = UINT64_MAX;
    #endif

    link = node->reversePeers;
    while (link) {
        struct Node_Link* nextLink = link->nextPeer;
        unlinkNodes(link, store);
        link = nextLink;
    }

    Assert_true(!node->bestParent);

    Assert_ifParanoid(node == RB_FIND(NodeRBTree, &store->nodeTree, node));
    RB_REMOVE(NodeRBTree, &store->nodeTree, node);
    store->pub.nodeCount--;

    Allocator_free(node->alloc);
}

struct Node_Link* NodeStore_discoverNode(struct NodeStore* nodeStore,
                                         struct Address* addr,
                                         struct EncodingScheme* scheme,
                                         int inverseLinkEncodingFormNumber,
                                         uint32_t reach)
{
    struct NodeStore_pvt* store = Identity_check((struct NodeStore_pvt*)nodeStore);
    verify(store);

    struct Node_Two* child = nodeForIp(store, addr->ip6.bytes);

    #ifdef Log_DEBUG
        uint8_t printedAddr[60];
        Address_print(printedAddr, addr);
        Log_debug(store->logger, "Discover node [%s]", printedAddr);
    #endif

    struct Allocator* alloc = NULL;
    if (!child) {
        alloc = Allocator_child(store->alloc);
        child = Allocator_calloc(alloc, sizeof(struct Node_Two), 1);
        child->alloc = alloc;
        Bits_memcpyConst(&child->address, addr, sizeof(struct Address));
        child->encodingScheme = EncodingScheme_clone(scheme, child->alloc);
        Identity_set(child);
    }

    // False if someone just updated.
    //Assert_true(child->address.protocolVersion);
    //Assert_true(EncodingScheme_equals(scheme, child->encodingScheme));//TODO

    struct Node_Link* link = discoverLink(store,
                                          store->selfLink,
                                          addr->path,
                                          child,
                                          addr->path,
                                          inverseLinkEncodingFormNumber);

    if (!link) {
        if (alloc) {
            Allocator_free(alloc);
        }
        verify(store);
        Log_debug(store->logger, "Invalid path");
        return NULL;
    }

    if (link->parent == store->pub.selfNode && !link->child->bestParent) {
        updateBestParent(link->child, link, reach, store);
    }

    handleNews(link->child, reach, store);
    freePendingLinks(store);

    while (store->pub.nodeCount >= store->pub.nodeCapacity
        || store->pub.linkCount >= store->pub.linkCapacity)
    {
        struct Node_Two* worst = getWorstNode(store);
        #ifdef Log_DEBUG
            uint8_t worstAddr[60];
            Address_print(worstAddr, &worst->address);
            Log_debug(store->logger, "store full, removing worst node: [%s] nodes [%d] links [%d]",
                      worstAddr, store->pub.nodeCount, store->pub.linkCount);
        #endif
        destroyNode(worst, store);
        freePendingLinks(store);
    }

    verify(store);
    return link;
}

struct Node_Two* NodeStore_nodeForAddr(struct NodeStore* nodeStore, uint8_t addr[16])
{
    struct NodeStore_pvt* store = Identity_check((struct NodeStore_pvt*)nodeStore);
    struct Node_Two* n = nodeForIp(store, addr);
    if (n && n->address.path == UINT64_MAX) {
        #ifdef Log_DEBUG
            uint8_t addrStr[40];
            AddrTools_printIp(addrStr, n->address.ip6.bytes);
            Log_debug(store->logger, "No way to represent path to [%s]", addrStr);
        #endif
        return NULL;
    }
    return n;
}

struct Node_Two* NodeStore_closestNode(struct NodeStore* nodeStore, uint64_t path)
{
    struct NodeStore_pvt* store = Identity_check((struct NodeStore_pvt*)nodeStore);
    struct Node_Link* out = NULL;
    findClosest(path, &out, NULL, store->selfLink, store);
    if (!out) { return NULL; }
    return Identity_check(out->child);
}

struct Node_Two* NodeStore_nodeForPath(struct NodeStore* nodeStore, uint64_t path)
{
    struct NodeStore_pvt* store = Identity_check((struct NodeStore_pvt*)nodeStore);
    struct Node_Link* out = NULL;
    uint64_t pathParentChild = findClosest(path, &out, NULL, store->selfLink, store);
    if (pathParentChild != 1) { return NULL; }
    return Identity_check(out->child);
}

struct Node_Link* NodeStore_getLinkOnPath(struct NodeStore* nodeStore,
                                          uint64_t routeLabel,
                                          uint32_t hopNum)
{
    struct NodeStore_pvt* store = Identity_check((struct NodeStore_pvt*)nodeStore);
    struct Node_Link* link = NULL;
    uint32_t num = hopNum;
    uint64_t path = findClosest(routeLabel, &link, &num, store->selfLink, store);
    if (path == findClosest_INVALID || num < hopNum) {
        return NULL;
    }
    return link;
}

struct Node_Link* NodeStore_getLink(struct Node_Two* parent, uint32_t linkNum)
{
    struct Node_Link* link = NULL;
    RB_FOREACH_REVERSE(link, PeerRBTree, &parent->peerTree) {
        if (!linkNum--) {
            return link;
        }
    }
    return NULL;
}

char* NodeStore_getRouteLabel_strerror(uint64_t returnVal)
{
    switch (returnVal) {
        case NodeStore_getRouteLabel_PARENT_NOT_FOUND:
            return "NodeStore_getRouteLabel_PARENT_NOT_FOUND";
        case NodeStore_getRouteLabel_CHILD_NOT_FOUND:
            return "NodeStore_getRouteLabel_CHILD_NOT_FOUND";
        default: return NULL;
    }
}

uint64_t NodeStore_getRouteLabel(struct NodeStore* nodeStore,
                                 uint64_t pathToParent,
                                 uint64_t pathParentToChild)
{
    struct NodeStore_pvt* store = Identity_check((struct NodeStore_pvt*)nodeStore);
    struct Node_Link* linkToParent;
    if (findClosest(pathToParent, &linkToParent, NULL, store->selfLink, store) != 1) {
        return NodeStore_getRouteLabel_PARENT_NOT_FOUND;
    }
    logLink(store, linkToParent, "NodeStore_getRouteLabel() PARENT");
    struct Node_Link* linkToChild = NULL;
    RB_FOREACH_REVERSE(linkToChild, PeerRBTree, &linkToParent->child->peerTree) {
        if (pathParentToChild == linkToChild->cannonicalLabel) {
            if (linkToParent == store->selfLink) {
                return linkToChild->cannonicalLabel;
            }
            return extendRoute(pathToParent,
                               linkToChild->parent->encodingScheme,
                               linkToChild->cannonicalLabel,
                               linkToParent->inverseLinkEncodingFormNumber);
        }
    }
    return NodeStore_getRouteLabel_CHILD_NOT_FOUND;
}

uint64_t NodeStore_optimizePath(struct NodeStore* nodeStore, uint64_t path)
{
    struct NodeStore_pvt* store = Identity_check((struct NodeStore_pvt*)nodeStore);
    struct Node_Link* linkToParent;
    uint64_t next = findClosest(path, &linkToParent, NULL, store->selfLink, store);
    if (next == findClosest_INVALID) {
        return NodeStore_optimizePath_INVALID;
    }
    if (linkToParent == store->selfLink) {
        if (next == 1) { return 1; }
        return path;
    }

    if (next == 1) { return linkToParent->child->address.path; }

    if (linkToParent->child->bestParent) {
        linkToParent = linkToParent->child->bestParent;
    }

    uint64_t optimized = extendRoute(linkToParent->child->address.path,
                                     linkToParent->child->encodingScheme,
                                     next,
                                     linkToParent->inverseLinkEncodingFormNumber);
    if (optimized < UINT64_MAX) {
        return optimized;
    }
    return path;
}

uint32_t NodeStore_linkCount(struct Node_Two* node)
{
    uint32_t i = 0;
    struct Node_Link* link;
    RB_FOREACH_REVERSE(link, PeerRBTree, &node->peerTree) {
        i++;
    }
    return i;
}

/** See: NodeStore.h */
struct NodeStore* NodeStore_new(struct Address* myAddress,
                                const uint32_t capacity,
                                struct Allocator* allocator,
                                struct Log* logger)
{
    struct Allocator* alloc = Allocator_child(allocator);

    struct NodeStore_pvt* out = Allocator_clone(alloc, (&(struct NodeStore_pvt) {
        .pub = {
            .nodeCapacity = NodeStore_DEFAULT_NODE_CAPACITY,
            .linkCapacity = NodeStore_DEFAULT_LINK_CAPACITY
        },
        .capacity = capacity,
        .logger = logger,
        .alloc = alloc
    }));
    Identity_set(out);

    // Create the self node
    struct Node_Two* selfNode = Allocator_calloc(alloc, sizeof(struct Node_Two), 1);
    Bits_memcpyConst(&selfNode->address, myAddress, sizeof(struct Address));
    selfNode->encodingScheme = NumberCompress_defineScheme(alloc);
    selfNode->alloc = alloc;
    Identity_set(selfNode);
    out->pub.selfNode = selfNode;
    selfNode->bestParent = linkNodes(selfNode, selfNode, 1, 0xffffffffu, 0, 1, out);
    selfNode->address.path = 1;
    setReach(selfNode, UINT32_MAX);
    out->selfLink = selfNode->reversePeers;
    RB_INSERT(NodeRBTree, &out->nodeTree, selfNode);

    out->pub.selfAddress = &out->selfLink->child->address;

    return &out->pub;
}


//////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////


/**
 * Dump the table, one node at a time.
 */
struct Node_Two* NodeStore_dumpTable(struct NodeStore* nodeStore, uint32_t index)
{
    struct NodeStore_pvt* store = Identity_check((struct NodeStore_pvt*)nodeStore);
    // TODO: Schlameil the painter
    uint32_t i = 0;
    struct Node_Two* nn = NULL;
    RB_FOREACH(nn, NodeRBTree, &store->nodeTree) {
        if (i++ == index) { return nn; }
    }
    return NULL;
}

static struct Node_Two* getBestCycleB(struct Node_Two* node,
                                      struct Address* target,
                                      struct NodeStore_pvt* store)
{
    uint32_t targetPfx = Address_getPrefix(target);
    uint32_t ourDistance = Address_getPrefix(store->pub.selfAddress) ^ targetPfx;
    struct Node_Link* next = NULL;
    RB_FOREACH_REVERSE(next, PeerRBTree, &node->peerTree) {
        if (next->child->bestParent != next || next == store->selfLink) { continue; }
        if (next->child->address.path == UINT64_MAX) { continue; }
        if ((Address_getPrefix(&next->child->address) ^ targetPfx) >= ourDistance) { continue; }
        return next->child;
    }

    return NULL;
}

static int getBestCycle(struct Node_Two* node,
                        struct Address* target,
                        struct Node_Two** output,
                        int limit,
                        int cycle,
                        struct NodeStore_pvt* store)
{
    Assert_always(cycle < 1000);
    if (cycle < limit) {
        int total = 0;
        struct Node_Link* next = NULL;
        RB_FOREACH_REVERSE(next, PeerRBTree, &node->peerTree) {
            if (*output) { return total; }
            if (next->child->bestParent != next || next == store->selfLink) { continue; }
            total += getBestCycle(next->child, target, output, limit, cycle+1, store);
        }
        return total;
    }

    *output = getBestCycleB(node, target, store);
    return 1;
}

struct Node_Two* NodeStore_getBest(struct Address* targetAddress, struct NodeStore* nodeStore)
{
    struct NodeStore_pvt* store = Identity_check((struct NodeStore_pvt*)nodeStore);
    struct Node_Two* n = NodeStore_nodeForAddr(nodeStore, targetAddress->ip6.bytes);
    if (n && n->bestParent) { return n; }

    for (int i = 0; i < 10000; i++) {
        int ret = getBestCycle(store->pub.selfNode, targetAddress, &n, i, 0, store);
        if (n || !ret) { return n; }
    }

    return NULL;
}

struct NodeList* NodeStore_getPeers(uint64_t label,
                                    const uint32_t max,
                                    struct Allocator* allocator,
                                    struct NodeStore* nodeStore)
{
    struct NodeStore_pvt* store = Identity_check((struct NodeStore_pvt*)nodeStore);

    // truncate the label to the part which this node uses...
    label &= Bits_maxBits64(NumberCompress_bitsUsedForLabel(label));

    struct NodeList* out = Allocator_calloc(allocator, sizeof(struct NodeList), 1);
    out->nodes = Allocator_calloc(allocator, sizeof(char*), max);

    struct Node_Link* next = NULL;
    RB_FOREACH_REVERSE(next, PeerRBTree, &store->pub.selfNode->peerTree) {
        uint64_t p = next->child->address.path;
        if (p > (((uint64_t)1)<<63)) { continue; }
        int j;
        for (j = 0; j < (int)max; j++) {
            if (out->nodes[j] && (out->nodes[j]->address.path ^ label) < (p ^ label)) {
                break;
            }
        }
        switch (j) {
            default: Bits_memmove(out->nodes, &out->nodes[1], (j - 1) * sizeof(char*));
            case 1: out->nodes[j - 1] = next->child;
            case 0:;
        }
    }

    out->size = 0;
    for (int i = 0; i < (int)max; i++) {
        if (out->nodes[i]) {
            out->nodes = &out->nodes[i];
            out->size = max - i;
            break;
        }
    }

    for (int i = 0; i < (int)out->size; i++) {
        Identity_check(out->nodes[i]);
        checkNode(out->nodes[i], store);
        Assert_true(out->nodes[i]->address.path);
        Assert_true(out->nodes[i]->address.path < (((uint64_t)1)<<63));
        out->nodes[i] = Allocator_clone(allocator, out->nodes[i]);
    }
    return out;
}

static bool isOkAnswer(struct Node_Two* node,
                       uint32_t compatVer,
                       struct NodeStore_pvt* store)
{
    if (node->address.path == UINT64_MAX) {
        // (very) unreachable
        return false;
    }
    if (!Version_isCompatible(compatVer, node->address.protocolVersion)) {
        return false;
    }
    if (node == store->pub.selfNode) {
        return false;
    }
    return true;
}

/** See: NodeStore.h */
struct NodeList* NodeStore_getClosestNodes(struct NodeStore* nodeStore,
                                           struct Address* targetAddress,
                                           const uint32_t count,
                                           uint32_t compatVer,
                                           struct Allocator* allocator)
{
    struct NodeStore_pvt* store = Identity_check((struct NodeStore_pvt*)nodeStore);

    struct NodeList* out = Allocator_malloc(allocator, sizeof(struct NodeList));
    out->nodes = Allocator_calloc(allocator, count, sizeof(char*));
    out->size = count;

    struct Node_Two fakeNode = { .marked = 0 };
    Bits_memcpyConst(&fakeNode.address, targetAddress, sizeof(struct Address));

    struct Node_Two* next = Identity_ncheck(RB_NFIND(NodeRBTree, &store->nodeTree, &fakeNode));
    if (!next) {
        out->size = 0;
        return out;
    }

    struct Node_Two* prev = Identity_ncheck(NodeRBTree_RB_PREV(next));
    int idx = out->size-1;

    while (idx > -1) {
        if (prev && (!next || Address_closest(targetAddress, &next->address, &prev->address) > 0)) {
            if (isOkAnswer(prev, compatVer, store)) { out->nodes[idx--] = prev; }
            prev = Identity_ncheck(NodeRBTree_RB_PREV(prev));
            continue;
        }
        if (next && (!prev || Address_closest(targetAddress, &next->address, &prev->address) < 0)) {
            if (isOkAnswer(next, compatVer, store)) { out->nodes[idx--] = next; }
            next = Identity_ncheck(NodeRBTree_RB_NEXT(next));
            continue;
        }
        break;
    }

    out->nodes = &out->nodes[idx+1];
    out->size -= idx+1;

    for (int i = 0; i < (int)out->size; i++) {
        Identity_check(out->nodes[i]);
        Assert_true(out->nodes[i]->address.path);
        Assert_true(out->nodes[i]->address.path < (((uint64_t)1)<<63));
        out->nodes[i] = Allocator_clone(allocator, out->nodes[i]);
    }
    return out;
}

void NodeStore_updateReach(struct NodeStore* nodeStore, struct Node_Two* node, uint32_t newReach)
{
    struct NodeStore_pvt* store = Identity_check((struct NodeStore_pvt*)nodeStore);
    handleNews(node, newReach, store);
}

void NodeStore_brokenPath(uint64_t path, struct NodeStore* nodeStore)
{
    struct NodeStore_pvt* store = Identity_check((struct NodeStore_pvt*)nodeStore);
    verify(store);
    #ifdef Log_DEBUG
        uint8_t pathStr[20];
        AddrTools_printPath(pathStr, path);
        Log_debug(store->logger, "NodeStore_brokenPath(%s)", pathStr);
    #endif
    struct Node_Two* nn = NodeStore_nodeForPath(nodeStore, path);
    if (nn && Node_getReach(nn) > 0) {
        handleBadNews(nn, 0, store);
    }
    verify(store);
}