myxce/src/controller/join.c

#include "config.h"

#include <errno.h>
#include <string.h>
#include <unistd.h>

#include "cluster.h"
#include "config_repl.h"
#include "ha.h"
#include "persist.h"
#include "xce/cluster_def.h"
#include "xce/errnox.h"

static int validate_host(cluster_t* c, const char* host, const char* key)
{
    return 1;
}

static int handle_client_join(cluster_t* c, endpoint_t* ep,
                              xce_request_t* request)
{
    xce_reply_t reply;

    init_reply_by_req(&reply, request);

    reply.status     = 0;
    reply.data_which = RequestReply_data_nodata;

    xce_reply(ep, c->myhost, &reply);

    return 0;
}

static int handle_worker_join(cluster_t* c, endpoint_t* ep,
                              xce_request_t* request)
{
    int               idx     = 0;
    ep_private_t*     priv    = (ep_private_t*)ep->data;
    xce_reply_t       reply;
    xce_peerinfo_t*   from    = request->from;
    int               rc      = 0;
    xce_joindata_t*   joinmsg = NULL;
    worker_t*         w       = NULL;
    xce_hostconfig_t* hconfig = NULL;

    TRACE("%s worker %s joining...", __func__, from->name);

    init_reply_by_req(&reply, request);

    if (!validate_host(c, from->name, from->key)) {
        ERROR("%s reject join of host %s", __func__, from->name);
        rc               = -EPERM;
        reply.status     = rc;
        reply.data_which = RequestReply_data_nodata;
        xce_reply(ep, c->myhost, &reply);
        return rc;
    }

    joinmsg = request->body.joinmsg;

    /* Fail fast before any lock: reject if no hostconfig matches (req 4.4) */
    hconfig = find_matched_hostconfig(c, from->name);
    if (hconfig == NULL) {
        rc               = (errno != 0) ? -errno : -EINVAL;
        reply.status     = rc;
        reply.data_which = RequestReply_data_nodata;
        xce_reply(ep, c->myhost, &reply);
        return rc;
    }

    /* BEFORE_EVENT fires before the existence check (req 5.1) */
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_WORKER_JOIN, c, from->name, (worker_t*)NULL);

    /* Explicit existence check under brief READ lock (req 1.1) */
    lock(c->workermgr, RTTI_LOCK_READ);
    rc = key_find(c->workermgr, (void*)from->name, (void**)&w);
    if (w != NULL)
        reference(w);
    unlock(c->workermgr);

    if (rc < 0 && rc != -ENOENT) {
        /* Non-ENOENT error: reject join (req 1.3) */
        ERROR("%s workermgr lookup error for %s, rc %d", __func__, from->name, rc);
        idx = rc;
    } else if (w == NULL) {
        /* Host not yet registered: first-time join (req 1.2, req 2) */
        idx = init_worker(c, ep, from->name, joinmsg->data.attrs, hconfig);
        if (idx < 0)
            ERROR("%s host %s first join failed, rc %d", __func__, from->name, -idx);
    } else {
        /* Host already registered: rejoin (req 1.2, req 3) */
        idx = rejoin_worker(c, w, ep, joinmsg->data.attrs, hconfig);
        dereference(w);
        w = NULL;
        if (idx < 0)
            ERROR("%s host %s rejoin failed, rc %d", __func__, from->name, -idx);
    }

    /* Look up worker for AFTER_EVENT persist hook (w is NULL on error) */
    if (idx >= 0)
        vget(c->workermgr, idx, (void**)&w);

    /* AFTER_EVENT fires after create/update, outside any lock (req 5.1) */
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_WORKER_JOIN, c, from->name, w);

    /* Commit barrier (ha-commit-barrier-rollout Req 1.1-1.4): the success
     * reply is sent only after the join's WAL entry is majority-committed;
     * barrier failures (-ETIMEDOUT, ...) become the error reply and skip
     * on_worker_join below. Join errors (idx < 0) reply immediately — no
     * barrier wait. No lock is held here (AFTER_EVENT runs outside any
     * lock). timeout 0 = configured sync_timeout_ms. */
    if (idx >= 0) {
        rc = persist_commit_barrier(&c->persist, 0);
        if (rc != 0) {
            WARN("%s commit barrier failed for %s, rc %d", __func__,
                 from->name, rc);
            idx = rc;
        }
    }

    reply.status     = idx;
    reply.data_which = RequestReply_data_nodata;
    xce_reply(ep, c->myhost, &reply);

    if (idx < 0)
        return idx;

    /* Set priv->u.idx before on_worker_join (req 2.3, req 5.2) */
    priv->u.idx = idx;
    on_worker_join(c, ep);

    return 0;
}

static int handle_scheduler_join(cluster_t* c, endpoint_t* ep,
                                 xce_request_t* request)
{
    xce_reply_t   reply;
    ep_private_t* priv = (ep_private_t*)ep->data;
    int           rc   = 0;

    TRACE("%s scheduler join, ep %p", __func__, ep);

    init_reply_by_req(&reply, request);

    c->sched.ep = ep;
    priv->u.idx = 0;

    rc = on_scheduler_join(c);

    reply.status     = rc;
    reply.data_which = RequestReply_data_nodata;
    xce_reply(ep, c->myhost, &reply);

    return rc;
}

static int handle_controller_join(cluster_t* c, endpoint_t* ep,
                                  xce_request_t* request)
{
    xce_reply_t reply;

    init_reply_by_req(&reply, request);
    reply.status     = 0;
    reply.data_which = RequestReply_data_nodata;
    xce_reply(ep, c->myhost, &reply);

    /* Register this endpoint for OP_upgradeConfig fan-out to standbys */
    if (c->ha.standby_endpoints) {
        ep_private_t *priv = (ep_private_t *)ep->data;
        if (priv)
            priv->role = HostRole_controller;

        pthread_mutex_lock(&c->ha.lock);
        zarray_store(c->ha.standby_endpoints, ep);
        pthread_mutex_unlock(&c->ha.lock);

        INFO("controller %s joined — %d standby(s) tracked",
             request->from ? request->from->name : "?",
             c->ha.standby_endpoints->count);

        /* Immediately push current config to the newly-joined standby so it
         * doesn't have to wait for the next scheduled push.  Transport-only
         * (no barrier wait — task 5.2 remediation r1): this is a best-effort
         * catch-up stream; convergence is owned by the OP_replConfig
         * anti-entropy pull and the heartbeat decision loop. */
        if (ha_is_primary(&c->ha))
            push_config_transport_only(c, NULL, 0, NULL, 0, NULL);
    }

    return 0;
}

static int handle_apiserver_join(cluster_t* c, endpoint_t* ep,
                                 xce_request_t* request)
{
    xce_reply_t     reply;
    xce_peerinfo_t* from = request->from;
    int             rc   = 0;

    TRACE("%s api server join, ep %p", __func__, ep);

    init_reply_by_req(&reply, request);

    rc = add_apiserver(c, from->name, ep);
    if (rc >= 0) {
        ep_private_t* priv = (ep_private_t*)ep->data;

        priv->u.idx = rc;
        on_apiserver_join(c, ep);
    }

    reply.status     = rc;
    reply.data_which = RequestReply_data_nodata;
    xce_reply(ep, c->myhost, &reply);

    if (rc < 0) {
        ERROR("api server %s join failed, rc %d", from->name, rc);
        return rc;
    }
    return 0;
}

static int handle_join_(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
    ep_private_t*   priv;
    int             rc = -EINVAL;
    int             role;
    xce_reply_t     reply;
    xce_peerinfo_t* from = request->from;

    init_reply_by_req(&reply, request);

    priv = (ep_private_t*)ep->data;
    if (priv == NULL) {
        rc = -E_XCE_INTERNAL;

        ERROR("priv data of endpoint should not be NULL");

        reply.status     = rc;
        reply.data_which = RequestReply_data_nodata;
        xce_reply(ep, c->myhost, &reply);

        return rc;
    }

    role       = request->body.joinmsg->role;
    priv->role = role;

    TRACE("%s role %d join from %s", __func__, role, from->name);

    /* ── HA guard (P1-6) ────────────────────────────────────────
     * A standby must never accept joins as if it were the primary.
     * If the primary address is known, tell the client where to go
     * (E_XCE_REDIRECT).  If no primary is known yet, tell the client
     * to retry later (E_XCE_NOMASTER).
     */
    if (ha_should_redirect(&c->ha)) {
        WARN("%s standby redirecting role-%d join from %s to %s:%u",
             __func__, role, from->name,
             c->ha.primary_host, c->ha.primary_port);
        rc               = -E_XCE_REDIRECT;
        reply.status     = rc;
        reply.data_which = RequestReply_data_nodata;
        xce_reply(ep, c->myhost, &reply);
        return rc;
    }
    if (ha_no_primary_known(&c->ha)) {
        WARN("%s standby has no primary yet, rejecting role-%d join from %s",
             __func__, role, from->name);
        rc               = -E_XCE_NOMASTER;
        reply.status     = rc;
        reply.data_which = RequestReply_data_nodata;
        xce_reply(ep, c->myhost, &reply);
        return rc;
    }

    switch (role) {
    case HostRole_client:
        rc = handle_client_join(c, ep, request);
        break;
    case HostRole_worker:
        rc = handle_worker_join(c, ep, request);
        break;
    case HostRole_scheduler:
        rc = handle_scheduler_join(c, ep, request);
        break;
    case HostRole_controller:
        rc = handle_controller_join(c, ep, request);
        break;
    case HostRole_apiserver:
        rc = handle_apiserver_join(c, ep, request);
        break;
    default:
        rc = -E_XCE_BAD_ROLE;
        ERROR("unknown role %d from host %s", role, from->name);

        reply.status     = rc;
        reply.data_which = RequestReply_data_nodata;
        xce_reply(ep, c->myhost, &reply);
        break;
    }

    return rc;
}

int handle_join(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
    int rc = 0;

    rc = handle_join_(c, ep, request);

    return rc;
}