static void sendMessageCallback(uv_write_t* uvReq, int error)
{
    struct Pipe_WriteRequest_pvt* req = Identity_check((struct Pipe_WriteRequest_pvt*) uvReq);
    if (error) {
        Log_info(req->pipe->pub.logger, "Failed to write to pipe [%s] [%s]",
                 req->pipe->pub.fullName, uv_strerror(error) );
    }
    req->pipe->queueLen -= req->msg->length;
    Assert_ifParanoid(req->pipe->queueLen >= 0);
    Allocator_free(req->alloc);
}

struct RouterModule_Promise* SearchRunner_search(uint8_t target[16],
                                                 int maxRequests,
                                                 int maxRequestsIfFound,
                                                 struct SearchRunner* searchRunner,
                                                 struct Allocator* allocator)
{
    struct SearchRunner_pvt* runner = Identity_check((struct SearchRunner_pvt*)searchRunner);

    if (runner->searches > runner->maxConcurrentSearches) {
        Log_debug(runner->logger, "Skipping search because there are already [%d] searches active",
                  runner->searches);
        return NULL;
    }

    if (maxRequests < 1) {
        maxRequests = SearchRunner_DEFAULT_MAX_REQUESTS;
    }
    if (maxRequestsIfFound < 1) {
        maxRequestsIfFound = SearchRunner_DEFAULT_MAX_REQUESTS_IF_FOUND;
    }

    struct Allocator* alloc = Allocator_child(allocator);

    struct Address targetAddr = { .path = 0 };
    Bits_memcpyConst(targetAddr.ip6.bytes, target, Address_SEARCH_TARGET_SIZE);

    struct NodeList* nodes =
        NodeStore_getClosestNodes(runner->nodeStore,
                                  &targetAddr,
                                  maxRequests,
                                  Version_CURRENT_PROTOCOL,
                                  alloc);

    if (nodes->size == 0) {
        Log_debug(runner->logger, "No nodes available for beginning search");
        Allocator_free(alloc);
        return NULL;
    }

    struct SearchStore_Search* sss = SearchStore_newSearch(target, runner->searchStore, alloc);

    for (int i = 0; i < (int)nodes->size; i++) {
        SearchStore_addNodeToSearch(&nodes->nodes[i]->address, sss);
    }

    struct SearchRunner_Search* search = Allocator_clone(alloc, (&(struct SearchRunner_Search) {
        .pub = {
            .alloc = alloc
        },
        .runner = runner,
        .search = sss,
        .maxRequests = maxRequests,
        .maxRequestsIfFound = maxRequestsIfFound
    }));

static int closeInterface(struct Allocator_OnFreeJob* job)
{
    struct Peer* toClose = Identity_check((struct Peer*) job->userData);

    sendPeer(0xffffffff, PFChan_Core_PEER_GONE, toClose);

    int index = Map_EndpointsBySockaddr_indexForHandle(toClose->handle, &toClose->ici->peerMap);
    Assert_true(index >= 0 && toClose->ici->peerMap.values[index] == toClose);
    Map_EndpointsBySockaddr_remove(index, &toClose->ici->peerMap);
    return 0;
}

static void addRemoveSomething(Dict* args,
                               void* vcontext,
                               String* txid,
                               struct Allocator* requestAlloc,
                               enum addRemoveSomething_What what)
{
    struct RouteGen_admin_Ctx* ctx = Identity_check((struct RouteGen_admin_Ctx*) vcontext);
    String* route = Dict_getString(args, String_CONST("route"));
    char* error = NULL;

    struct Sockaddr_storage ss;
    if (route->len > 63) {
        error = "parse_failed";
    }
    if (!error) {
        if (Sockaddr_parse(route->bytes, &ss)) {
            error = "parse_failed";
        } else {
            int family = Sockaddr_getFamily(&ss.addr);
            if (family != Sockaddr_AF_INET && family != Sockaddr_AF_INET6) {
                error = "unexpected_af";
            }
        }
    }
    int retVal = -1;
    Dict* out = Dict_new(requestAlloc);
    if (!error) {
        switch (what) {
            case addRemoveSomething_What_ADD_EXCEPTION:
                RouteGen_addException(ctx->rg, &ss.addr); break;
            case addRemoveSomething_What_ADD_PREFIX:
                RouteGen_addPrefix(ctx->rg, &ss.addr); break;
            case addRemoveSomething_What_ADD_LOCALPREFIX:
                RouteGen_addLocalPrefix(ctx->rg, &ss.addr); break;
            case addRemoveSomething_What_RM_EXCEPTION:
                retVal = RouteGen_removeException(ctx->rg, &ss.addr); break;
            case addRemoveSomething_What_RM_PREFIX:
                retVal = RouteGen_removePrefix(ctx->rg, &ss.addr); break;
            case addRemoveSomething_What_RM_LOCALPREFIX:
                retVal = RouteGen_removeLocalPrefix(ctx->rg, &ss.addr); break;
            default: Assert_failure("invalid op");
        }
        if (!retVal) {
            error = "no_such_route";
        } else {
            error = "none";
        }
    }
    Dict_putString(out,
                   String_new("error", requestAlloc),
                   String_new(error, requestAlloc),
                   requestAlloc);
    Admin_sendMessage(out, txid, ctx->admin);
}

// Incoming message which has passed through the cryptoauth and needs to be forwarded to the switch.
static Iface_DEFUN receivedPostCryptoAuth(struct Message* msg,
                                          struct Peer* ep,
                                          struct InterfaceController_pvt* ic)
{
    ep->bytesIn += msg->length;

    int caState = CryptoAuth_getState(ep->caSession);
    if (ep->state < InterfaceController_PeerState_ESTABLISHED) {
        // EP states track CryptoAuth states...
        ep->state = caState;
        SwitchCore_setInterfaceState(&ep->switchIf, SwitchCore_setInterfaceState_ifaceState_UP);

        Bits_memcpyConst(ep->addr.key, ep->caSession->herPublicKey, 32);
        Address_getPrefix(&ep->addr);

        if (caState == CryptoAuth_ESTABLISHED) {
            moveEndpointIfNeeded(ep);
            sendPeer(0xffffffff, PFChan_Core_PEER, ep);
        } else {
            // prevent some kinds of nasty things which could be done with packet replay.
            // This is checking the message switch header and will drop it unless the label
            // directs it to *this* router.
            if (msg->length < 8 || msg->bytes[7] != 1) {
                Log_info(ic->logger, "DROP message because CA is not established.");
                return 0;
            } else {
                // When a "server" gets a new connection from a "client" the router doesn't
                // know about that client so if the client sends a packet to the server, the
                // server will be unable to handle it until the client has sent inter-router
                // communication to the server. Here we will ping the client so when the
                // server gets the ping response, it will insert the client into its table
                // and know its version.

                // prevent DoS by limiting the number of times this can be called per second
                // limit it to 7, this will affect innocent packets but it doesn't matter much
                // since this is mostly just an optimization and for keeping the tests happy.
                if ((ep->pingCount + 1) % 7) {
                    sendPing(ep);
                }
            }
        }
    } else if (ep->state == InterfaceController_PeerState_UNRESPONSIVE
        && caState == CryptoAuth_ESTABLISHED)
    {
        ep->state = InterfaceController_PeerState_ESTABLISHED;
        SwitchCore_setInterfaceState(&ep->switchIf, SwitchCore_setInterfaceState_ifaceState_UP);
    } else {
        ep->timeOfLastMessage = Time_currentTimeMilliseconds(ic->eventBase);
    }

    Identity_check(ep);
    Assert_true(!(msg->capacity % 4));
    return Iface_next(&ep->switchIf, msg);
}

static void dumpTable(Dict* args, void* vcontext, String* txid, struct Allocator* requestAlloc)
{
    struct Context* ctx = Identity_check((struct Context*) vcontext);
    int64_t* page = Dict_getInt(args, String_CONST("page"));
    int ctr = (page) ? *page * ENTRIES_PER_PAGE : 0;

    Dict* out = Dict_new(requestAlloc);
    List* table = List_new(requestAlloc);
    struct Node_Two* nn = NULL;
    for (int i = 0; i < ctr+ENTRIES_PER_PAGE; i++) {
        nn = NodeStore_getNextNode(ctx->store, nn);
        if (!nn) { break; }
        if (i < ctr) { continue; }
        Dict* nodeDict = Dict_new(requestAlloc);

        String* ip = String_newBinary(NULL, 39, requestAlloc);
        Address_printIp(ip->bytes, &nn->address);
        Dict_putString(nodeDict, String_CONST("ip"), ip, requestAlloc);

        String* addr = Address_toString(&nn->address, requestAlloc);
        Dict_putString(nodeDict, String_CONST("addr"), addr, requestAlloc);

        String* path = String_newBinary(NULL, 19, requestAlloc);
        AddrTools_printPath(path->bytes, nn->address.path);
        Dict_putString(nodeDict, String_CONST("path"), path, requestAlloc);

        Dict_putInt(nodeDict, String_CONST("link"), Node_getCost(nn), requestAlloc);
        Dict_putInt(nodeDict, String_CONST("version"), nn->address.protocolVersion, requestAlloc);

        Dict_putInt(nodeDict,
                    String_CONST("time"),
                    NodeStore_timeSinceLastPing(ctx->store, nn),
                    requestAlloc);

        Dict_putInt(nodeDict,
                    String_CONST("bucket"),
                    NodeStore_bucketForAddr(ctx->store->selfAddress, &nn->address),
                    requestAlloc);

        List_addDict(table, nodeDict, requestAlloc);
    }
    Dict_putList(out, String_CONST("routingTable"), table, requestAlloc);

    if (nn) {
        Dict_putInt(out, String_CONST("more"), 1, requestAlloc);
    }
    Dict_putInt(out, String_CONST("count"), ctx->store->nodeCount, requestAlloc);
    Dict_putInt(out, String_CONST("peers"), ctx->store->peerCount, requestAlloc);

    Dict_putString(out, String_CONST("deprecation"),
        String_CONST("ip,path,version will soon be removed"), requestAlloc);

    Admin_sendMessage(out, txid, ctx->admin);
}

List* CryptoAuth_getUsers(struct CryptoAuth* context, struct Allocator* alloc)
{
    struct CryptoAuth_pvt* ca = Identity_check((struct CryptoAuth_pvt*) context);

    List* users = List_new(alloc);
    for (struct CryptoAuth_User* u = ca->users; u; u = u->next) {
        List_addString(users, String_clone(u->login, alloc), alloc);
    }

    return users;
}

static void checkPermissions(Dict* args, void* vctx, String* txid, struct Allocator* requestAlloc)
{
    struct Context* const ctx = Identity_check((struct Context*) vctx);
    struct Jmp jmp;
    Jmp_try(jmp) {
        checkPermissionsB(&jmp.handler, txid, ctx->admin, requestAlloc);
    } Jmp_catch {
        sendError(jmp.message, txid, ctx->admin);
        return;
    }
}

static void onPingResponse(struct SwitchPinger_Response* resp, void* onResponseContext)
{
    if (SwitchPinger_Result_OK != resp->res) {
        return;
    }
    struct Peer* ep = Identity_check((struct Peer*) onResponseContext);
    struct InterfaceController_pvt* ic = Identity_check(ep->ici->ic);

    ep->addr.protocolVersion = resp->version;

    if (Defined(Log_DEBUG)) {
        String* addr = Address_toString(&ep->addr, resp->ping->pingAlloc);
        if (!Version_isCompatible(Version_CURRENT_PROTOCOL, resp->version)) {
            Log_debug(ic->logger, "got switch pong from node [%s] with incompatible version",
                                  addr->bytes);
        } else if (ep->addr.path != resp->label) {
            uint8_t sl[20];
            AddrTools_printPath(sl, resp->label);
            Log_debug(ic->logger, "got switch pong from node [%s] mismatch label [%s]",
                                  addr->bytes, sl);
        } else {
            Log_debug(ic->logger, "got switch pong from node [%s]", addr->bytes);
        }
    }

    if (!Version_isCompatible(Version_CURRENT_PROTOCOL, resp->version)) {
        return;
    }

    if (ep->state == InterfaceController_PeerState_ESTABLISHED) {
        sendPeer(0xffffffff, PFChan_Core_PEER, ep);
    }

    ep->timeOfLastPing = Time_currentTimeMilliseconds(ic->eventBase);

    if (Defined(Log_DEBUG)) {
        String* addr = Address_toString(&ep->addr, resp->ping->pingAlloc);
        Log_debug(ic->logger, "Received [%s] from lazy endpoint [%s]",
                  SwitchPinger_resultString(resp->res)->bytes, addr->bytes);
    }
}

static uint8_t messageOut(struct Message* msg, struct Interface* iface)
{
    struct Context* ctx = Identity_check((struct Context*) iface->receiverContext);
    Assert_true(ctx->currentMessage < ctx->messageCount);

    // The list is populated backwards so we have to count down...
    struct Message* ctrlMsg = ctx->messages[ctx->messageCount - (++ctx->currentMessage)];

    Assert_true(ctrlMsg->length == msg->length);
    Assert_true(!Bits_memcmp(ctrlMsg->bytes, msg->bytes, msg->length));
    return 0;
}

static void sessionStats(Dict* args,
                         void* vcontext,
                         String* txid,
                         struct Allocator* alloc)
{
    struct Context* context = Identity_check((struct Context*) vcontext);
    int64_t* handleP = Dict_getInt(args, String_CONST("handle"));
    uint32_t handle = *handleP;

    struct SessionManager_Session* session = SessionManager_sessionForHandle(handle, context->sm);

    Dict* r = Dict_new(alloc);
    if (!session) {
        Dict_putString(r, String_CONST("error"), String_CONST("no such session"), alloc);
        Admin_sendMessage(r, txid, context->admin);
        return;
    }

    uint8_t printedAddr[40];
    AddrTools_printIp(printedAddr, session->caSession->herIp6);
    Dict_putString(r, String_CONST("ip6"), String_new(printedAddr, alloc), alloc);

    String* state =
        String_new(CryptoAuth_stateString(CryptoAuth_getState(session->caSession)), alloc);
    Dict_putString(r, String_CONST("state"), state, alloc);

    struct ReplayProtector* rp = &session->caSession->replayProtector;
    Dict_putInt(r, String_CONST("duplicates"), rp->duplicates, alloc);
    Dict_putInt(r, String_CONST("lostPackets"), rp->lostPackets, alloc);
    Dict_putInt(r, String_CONST("receivedOutOfRange"), rp->receivedOutOfRange, alloc);

    struct Address addr;
    Bits_memcpyConst(addr.key, session->caSession->herPublicKey, 32);
    addr.path = session->sendSwitchLabel;
    addr.protocolVersion = session->version;

    Dict_putString(r, String_CONST("addr"), Address_toString(&addr, alloc), alloc);

    Dict_putString(r, String_CONST("publicKey"),
                      Key_stringify(session->caSession->herPublicKey, alloc), alloc);
    Dict_putInt(r, String_CONST("version"), session->version, alloc);
    Dict_putInt(r, String_CONST("handle"), session->receiveHandle, alloc);
    Dict_putInt(r, String_CONST("sendHandle"), session->sendHandle, alloc);

    Dict_putInt(r, String_CONST("timeOfLastIn"), session->timeOfLastIn, alloc);
    Dict_putInt(r, String_CONST("timeOfLastOut"), session->timeOfLastOut, alloc);

    Dict_putString(r, String_CONST("deprecation"),
        String_CONST("publicKey,version will soon be removed"), alloc);

    Admin_sendMessage(r, txid, context->admin);
    return;
}

int SupernodeHunter_listSnodes(struct SupernodeHunter* snh,
                               struct Address*** outP,
                               struct Allocator* alloc)
{
    struct SupernodeHunter_pvt* snp = Identity_check((struct SupernodeHunter_pvt*) snh);
    struct Address** out = Allocator_calloc(alloc, sizeof(char*), snp->authorizedSnodes->length);
    for (int i = 0; i < snp->authorizedSnodes->length; i++) {
        out[i] = AddrSet_get(snp->authorizedSnodes, i);
    }
    *outP = out;
    return snp->authorizedSnodes->length;
}

static uint8_t sendMessage(struct Message* message, struct Interface* iface)
{
    struct TUNInterface_Illumos_pvt* ctx = Identity_check((struct TUNInterface_Illumos_pvt*)iface);

    Message_shift(message, -4);
    uint16_t ethertype = ((uint16_t*) message->bytes)[-1];
    if (ethertype != Ethernet_TYPE_IP6 && ethertype != Ethernet_TYPE_IP4) {
        Assert_true(!"Unsupported ethertype");
    }

    return Interface_sendMessage(&ctx->pipe->iface, message);
}

struct Allocator* Allocator_getChild(struct Allocator* alloc, int childNumber)
{
    struct Allocator_pvt* ctx = Identity_check((struct Allocator_pvt*)alloc);
    if (childNumber < 0) {
        return NULL;
    }
    struct Allocator_pvt* child = ctx->firstChild;
    for (;child && childNumber > 0; childNumber--) {
        child = child->nextSibling;
    }
    return (child) ? &child->pub : NULL;
}

struct Allocator_Allocation* Allocator_getAllocation(struct Allocator* alloc, int allocNum)
{
    struct Allocator_pvt* ctx = Identity_check((struct Allocator_pvt*)alloc);
    if (allocNum < 0) {
        return NULL;
    }
    struct Allocator_Allocation_pvt* allocation = ctx->allocations;
    for (;allocation && allocNum > 0; allocNum--) {
        allocation = allocation->next;
    }
    return (allocation) ? &allocation->pub : NULL;
}

static int searchOnFree(struct Allocator_OnFreeJob* job)
{
    struct SearchRunner_Search* search =
        Identity_check((struct SearchRunner_Search*)job->userData);

    *search->thisSearch = search->nextSearch;
    if (search->nextSearch) {
        search->nextSearch->thisSearch = search->thisSearch;
    }
    Assert_true(search->runner->searches > 0);
    search->runner->searches--;
    return 0;
}

static void chroot(Dict* args, void* vctx, String* txid, struct Allocator* requestAlloc)
{
    struct Context* const ctx = Identity_check((struct Context*) vctx);
    struct Jmp jmp;
    Jmp_try(jmp) {
        String* root = Dict_getStringC(args, "root");
        Security_chroot(root->bytes, &jmp.handler);
    } Jmp_catch {
        sendError(jmp.message, txid, ctx->admin);
        return;
    }
    sendError("none", txid, ctx->admin);
}

void EventEmitter_regCore(struct EventEmitter* eventEmitter,
                          struct Iface* iface,
                          enum PFChan_Pathfinder ev)
{
    struct EventEmitter_pvt* ee = Identity_check((struct EventEmitter_pvt*) eventEmitter);
    iface->connectedIf = &ee->trickIf;
    struct ArrayList_Ifaces* l = getHandlers(ee, ev, true);
    if (!l) {
        Assert_true(ev == 0);
        return;
    }
    ArrayList_Ifaces_add(l, iface);
}

void Allocator__adopt(struct Allocator* adoptedParent,
                      struct Allocator* childToAdopt,
                      const char* file,
                      int line)
{
    struct Allocator_pvt* parent = Identity_check((struct Allocator_pvt*) adoptedParent);
    struct Allocator_pvt* child = Identity_check((struct Allocator_pvt*) childToAdopt);

    if (parent->pub.isFreeing) { return; }
    Assert_true(!child->pub.isFreeing);

    if (isAncestorOf(child, parent)) {
        // The child is a parent of the parent, this means an adoption would be meaningless
        // because if the child is otherwise freed, it will take the parent along with it.
        return;
    }

    if (!parent->adoptions) {
        parent->adoptions =
            Allocator__calloc(adoptedParent, sizeof(struct Allocator_Adoptions), 1, file, line);
    }
    if (!child->adoptions) {
        child->adoptions =
            Allocator__calloc(childToAdopt, sizeof(struct Allocator_Adoptions), 1, file, line);
    }

    struct Allocator_List* pl =
        Allocator__calloc(childToAdopt, sizeof(struct Allocator_List), 1, file, line);
    pl->alloc = child;
    pl->next = parent->adoptions->children;
    parent->adoptions->children = pl;

    struct Allocator_List* cl =
        Allocator__calloc(childToAdopt, sizeof(struct Allocator_List), 1, file, line);
    cl->alloc = parent;
    cl->next = child->adoptions->parents;
    child->adoptions->parents = cl;
}

static void onBestPathChange(void* vPathfinder, struct Node_Two* node)
{
    struct Pathfinder_pvt* pf = Identity_check((struct Pathfinder_pvt*) vPathfinder);
    struct Allocator* alloc = Allocator_child(pf->alloc);
    if (pf->bestPathChanges > 128) {
        String* addrPrinted = Address_toString(&node->address, alloc);
        Log_debug(pf->log, "Ignore best path change from NodeStore [%s]", addrPrinted->bytes);
    } else {
        pf->bestPathChanges++;
        struct Message* msg = Message_new(0, 256, alloc);
        Iface_CALL(sendNode, msg, &node->address, Node_getCost(node), pf);
    }
    Allocator_free(alloc);
}