myxce/src/controller/worker.c

#include <errno.h>
#include <stdbool.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#include "cluster.h"
#include "controller.capnp.h"
#include "persist.h"
#include "wobj.h"
#include "worker.h"
#include "xce/cluster_def.h"
#include "xce/errnox.h"
#include "xstring.h"

static int hostmatched(const char* pattern, const char* name)
{
    if (strcmp(pattern, name) == 0) {
        return 1;
    }

    return 0;
}

xce_hostconfig_t* find_matched_hostconfig(cluster_t* c, const char* name)
{
    xce_hostconfig_t*  defaultconfig = NULL;
    xce_hostconfig_t** hconfigs      = NULL;
    int                i, nhconfigs;

    if (c->myconf == NULL) {
        errno = EINVAL;
        return NULL;
    }

    if (c->myconf->cluster_config == NULL) {
        errno = EINVAL;
        return NULL;
    }

    if (c->myconf->cluster_config->hconfigs == NULL) {
        return NULL;
    }

    nhconfigs = c->myconf->cluster_config->nhconfigs;
    hconfigs  = c->myconf->cluster_config->hconfigs;

    for (i = 0; i < nhconfigs; i++) {
        xce_hostconfig_t* config = hconfigs[i];

        if (strcmp(config->name, "default") == 0) {
            defaultconfig = config;
            continue;
        }

        if (hostmatched(config->name, name)) {
            return config;
        }
    }

    return defaultconfig;
}

static int update_val_res(cluster_t* c, worker_t* w, xce_resource_t* r,
                          xce_val_t* val)
{
    int        idx;
    zarray_t*  array = NULL;
    xce_val_t* v;

    idx = r->idx;

    if (r->flags & XCE_RESF_SHARED) {
        /* update shared resource */
        array = &(c->sharedinst);
    } else {
        /* update host based resource */
        array = &(w->host.resvals);
    }

    if (idx < array->caps) {
        v = (xce_val_t*)array->data[idx];
    } else {
        v = NULL;
    }

    if (v == NULL) {
        int rc;

        v  = (xce_val_t*)calloc(1, sizeof(*v));
        rc = zarray_put(array, idx, v);
        if (rc < 0) {
            FREEUP(v);
            return rc;
        }
    }

    memcpy(v, val, sizeof(*v));

    /* bug-1568 instrumentation: the actual VALUE being written into
     * w->host.resvals[idx] (or c->sharedinst[idx] for a shared resource)
     * -- earlier instrumentation (core_events.c/snapshot_wire.c/
     * snapshot_load.c/sched.c) only ever counted entries flowing through
     * the pipeline, never inspected their CONTENT; this is the true
     * source write, so a 0.0 logged here for e.g. 'slots'/'maxmem' means
     * the zero is genuine at the point of writing, not lost downstream. */
    if (val->which == Value_u_fvalue) {
        TRACE("%s resource '%s' worker '%s' idx %d -> fvalue=%f", __func__,
              r->name, (r->flags & XCE_RESF_SHARED) ? "(shared)"
                                                     : w->host.name,
              idx, val->u.fvalue);
    } else {
        TRACE("%s resource '%s' worker '%s' idx %d -> which=%d (non-numeric)",
              __func__, r->name,
              (r->flags & XCE_RESF_SHARED) ? "(shared)" : w->host.name, idx,
              val->which);
    }

    return 0;
}

static int update_resval(cluster_t* c, worker_t* w, xce_resval_t* val)
{
    xce_resource_t* r = NULL;
    xce_val_t*      v = NULL;
    int             rc;

    lock(c->resmgr, RTTI_LOCK_READ);
    rc = key_find(c->resmgr, (void*)val->name, (void**)&r);
    unlock(c->resmgr);

    if (rc < 0) {
        return rc;
    }
    if (r == NULL) {
        ERROR("resource `%s' can not be found", val->name);
        return -ENOENT;
    }

    if (r->idx < 0) {
        ERROR("invalid resource definition for %s", val->name);
        return -EINVAL;
    }

    v = val->val;

    rc = update_val_res(c, w, r, v);

    return rc;
}

/*
 * update_resval() returns its failure reason as a raw negative code
 * directly (key_find/zarray_put convention throughout this codebase --
 * never via errno), so logging it with PERROR (which reads the process-
 * global errno) reported whatever errno happened to be left over from an
 * unrelated earlier call -- "Success" if nothing had touched it, or a
 * stray EAGAIN/etc from some other syscall in the same event-loop
 * iteration, never the real reason. Decode the actual rc instead.
 */
static void log_update_resval_failure(const char* name, int rc)
{
    char ebuf[128];

    ebuf[0] = '\0';
    xce_strerror_r(-rc, ebuf, sizeof(ebuf));
    ERROR("update_resval `%s': %s", name, ebuf);
}

int update_resvals(cluster_t* c, worker_t* w, int nres, xce_resval_t* vals)
{
    int rc = 0;
    int i;

    for (i = 0; i < nres; i++) {
        rc = update_resval(c, w, &(vals[i]));
        if (rc < 0) {
            log_update_resval_failure(vals[i].name, rc);
        }
    }

    return 0;
}

int update_resval_list(cluster_t* c, worker_t* w, xce_resval_list_t* l)
{
    int rc = 0;
    int i;

    for (i = 0; i < l->count; i++) {
        rc = update_resval(c, w, l->vals[i]);
        if (rc < 0) {
            log_update_resval_failure(l->vals[i]->name, rc);
        }
    }

    return 0;
}

static int mark_host_res(cluster_t* c, worker_t* w, xce_config_t* config)
{
    xce_resource_t* r = NULL;
    char*           n = config->name;
    int             rc;
    /* clang-format-off */
    xce_val_t val;
    /* clang-format-on */

    lock(c->resmgr, RTTI_LOCK_READ);
    rc = key_find(c->resmgr, (void*)n, (void**)&r);
    unlock(c->resmgr);

    if (rc < 0) {
        return rc;
    }

    if (r == NULL) {
        ERROR("resource `%s' can not be found", n);
        return -ENOENT;
    }

    if (r->idx < 0) {
        ERROR("invalid resource definition for `%s'", n);
        return -EINVAL;
    }

    if (r->flags & XCE_RESF_SHARED) {
        ERROR("resource `%s' should not be a shared resource", n);
        return -EINVAL;
    }

    switch (r->valtype) {
    case XCE_VAL_STRING:
        val.which = Value_u_svalue;
        if (config->value == NULL) {
            val.u.svalue = "";
        } else {
            val.u.svalue = config->value;
        }

        break;
    case XCE_VAL_NUMERIC:
        val.which = Value_u_fvalue;
        if (config->value == NULL) {
            val.u.fvalue = 0.0;
        } else {
            val.u.fvalue = atof(config->value);
        }
        break;
    case XCE_VAL_BOOLEAN:
        val.which    = Value_u_bvalue;
        val.u.bvalue = 1;
        break;
    }
    rc = update_val_res(c, w, r, &val);

    return rc;
}

static int add_hostattrs_resvals(cluster_t* c, worker_t* w)
{
#define N_HOST_ATTRS (8)
    xce_val_t values[N_HOST_ATTRS] = {
        { .which = Value_u_svalue, .u = { .svalue = w->host.attrs->hosttype } },
        { .which = Value_u_svalue,
         .u     = { .svalue = w->host.attrs->hostmodel }                      },
        { .which = Value_u_fvalue,
         .u     = { .fvalue = w->host.attrs->cpufactor }                      },
        { .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->maxcpus }  },
        { .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->maxmem }   },
        { .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->maxswap }  },
        { .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->maxtmp }   },
        { .which = Value_u_fvalue, .u = { .fvalue = w->host.attrs->ndisks }   }
    };
    xce_resval_t resvals[N_HOST_ATTRS] = {
        { .name = "type",    .val = &values[0] },
        { .name = "model",   .val = &values[1] },
        { .name = "cpuf",    .val = &values[2] },
        { .name = "maxcpus", .val = &values[3] },
        { .name = "maxmem",  .val = &values[4] },
        { .name = "maxswap", .val = &values[5] },
        { .name = "maxtmp",  .val = &values[6] },
        { .name = "ndisks",  .val = &values[7] }
    };

    return update_resvals(c, w, N_HOST_ATTRS, &(resvals[0]));
}

static int set_host_slots(cluster_t* c, worker_t* w, int nslots)
{
    /* clang-format off */
    xce_val_t values[1] = {
        {
            .which = Value_u_fvalue,
            .u = {
                .fvalue = (float)nslots
            }
        }
    };
    xce_resval_t resvals[1] = {
        {
            .name = "slots",
            .val = &values[0]
        }
    };
    /* clang-format on */

    return update_resvals(c, w, 1, &(resvals[0]));
}

/* Bits set by admin commands that must survive a worker rejoin */
#define XCE_WORKER_ADMIN_MASK \
    (XCE_WORKER_LOCKEDU | XCE_WORKER_CLOSED | XCE_WORKER_HIDDEN \
     | XCE_WORKER_CLOSED_BY_ADMIN)

/*
 * apply_join_state — shared status/resource update for both create and rejoin.
 * Precondition: caller holds per-worker WRITE lock.
 */
#ifdef XCE_TESTING
int apply_join_state(cluster_t* c, worker_t* w, endpoint_t* ep,
                     xce_hostconfig_t* hconfig)
#else
static int apply_join_state(cluster_t* c, worker_t* w, endpoint_t* ep,
                            xce_hostconfig_t* hconfig)
#endif
{
    int nslots;
    int i;

    w->host.status &= ~XCE_WORKER_UNAVAIL;
    w->ep = ep;

    if (strcmp(hconfig->hosttype, "!") != 0) {
        dereference(w->host.attrs->hosttype);
        w->host.attrs->hosttype = makeString(hconfig->hosttype);
    }

    if (strcmp(hconfig->hostmodel, "!") != 0) {
        dereference(w->host.attrs->hostmodel);
        w->host.attrs->hostmodel = makeString(hconfig->hostmodel);
    }

    add_hostattrs_resvals(c, w);

    nslots = hconfig->mxj;
    if (nslots < 0)
        nslots = w->host.attrs->maxcpus;
    set_host_slots(c, w, nslots);

    if (w->host.dispatch_window_config != NULL) {
        free_timewindow_config(w->host.dispatch_window_config);
        w->host.dispatch_window_config = NULL;
    }
    if (hconfig->dispatchwindow != NULL)
        w->host.dispatch_window_config
            = dup_timewindow_config(hconfig->dispatchwindow);

    if (w->host.thresholds != NULL) {
        free_resource_thresholds(w->host.thresholds, w->host.nthresholds);
        w->host.nthresholds = 0;
        w->host.thresholds  = NULL;
    }
    if (hconfig->thresholds != NULL) {
        w->host.thresholds  = dup_resource_thresholds(hconfig->thresholds,
                                                      hconfig->nthresholds);
        w->host.nthresholds = hconfig->nthresholds;
    }

    for (i = 0; i < hconfig->nresources; i++)
        mark_host_res(c, w, hconfig->resources[i]);

    return 0;
}

int init_worker(cluster_t* c, endpoint_t* ep, const char* host,
                hostattrs_t* attrs, xce_hostconfig_t* hconfig)
{
    xce_worker_t tmp;
    worker_t*    w;
    int          rc;

    memset(&tmp, 0, sizeof(tmp));
    tmp.name  = (char*)host;
    tmp.attrs = attrs;

    w = makeWorker(c->workermgr, &tmp);
    if (w == NULL) {
        ERROR("%s makeWorker failed for %s", __func__, host);
        return -errno;
    }

    lock(w, RTTI_LOCK_WRITE);
    rc = apply_join_state(c, w, ep, hconfig);
    unlock(w);

    if (rc < 0)
        return rc;

    return w->idx;
}

int rejoin_worker(cluster_t* c, worker_t* w, endpoint_t* ep,
                  hostattrs_t* attrs, xce_hostconfig_t* hconfig)
{
    int admin_status;
    int rc;

    lock(w, RTTI_LOCK_WRITE);

    admin_status = w->host.status & XCE_WORKER_ADMIN_MASK;

    dereference(w->host.attrs->hosttype);
    w->host.attrs->hosttype = makeString(attrs->hosttype);
    dereference(w->host.attrs->hostmodel);
    w->host.attrs->hostmodel = makeString(attrs->hostmodel);

    w->host.attrs->cpufactor = attrs->cpufactor;
    w->host.attrs->maxcpus   = attrs->maxcpus;
    w->host.attrs->maxmem    = attrs->maxmem;
    w->host.attrs->maxswap   = attrs->maxswap;
    w->host.attrs->maxtmp    = attrs->maxtmp;
    w->host.attrs->ndisks    = attrs->ndisks;

    rc = apply_join_state(c, w, ep, hconfig);

    w->host.status |= admin_status;

    unlock(w);

    if (rc < 0)
        return rc;

    return w->idx;
}


static void reset_hostattrs_unknown(hostattrs_t* a)
{
    dereference(a->hosttype);
    a->hosttype  = makeString("unknown");
    dereference(a->hostmodel);
    a->hostmodel = makeString("unknown");
    a->cpufactor = 0;
    a->maxcpus   = 0;
    a->maxmem    = 0;
    a->maxswap   = 0;
    a->maxtmp    = 0;
    a->ndisks    = 0;
}

int restore_worker(void* cls, xce_worker_t host)
{
    cluster_t* c   = (cluster_t*)cls;
    int        idx = 0;
    worker_t*  w   = NULL;

    w = makeWorker(c->workermgr, &host);
    if (w == NULL) {
        return -errno;
    }

    lock(w, RTTI_LOCK_WRITE);

    if (w->host.attrs != NULL)
        reset_hostattrs_unknown(w->host.attrs);
    add_hostattrs_resvals(c, w);
    idx = w->idx;
    w->host.status |= XCE_WORKER_UNAVAIL;

    unlock(w);

    return idx;
}

/*
 * restore_host_state — restore a host admin-state frame (config-upgrade-
 * snapshot Requirements 1.10, 1.11, 5.8).
 *
 * Unlike restore_queue_state() (queues are config-provisioned, so an
 * update-only lookup is always safe), a worker_t is join-provisioned: it
 * may not exist yet on a member that hasn't locally replayed this host's
 * original join record (e.g. one that received the upgrade snapshot via
 * the out-of-band install path). So this restore path is create-if-missing:
 *
 *   - Hit (host already registered -- a real joined worker, or an earlier
 *     placeholder left by a prior restore attempt): OR the restored admin
 *     bits into the existing status only. Nothing else is touched, so a
 *     real worker's live attrs/ep/resource values survive untouched.
 *   - Miss: create a minimal placeholder via makeWorker(), mirroring the
 *     synthetic lost_and_found sentinel host precedent exactly
 *     (config_repl.c:1114-1130) -- the same "makeWorker() without a join"
 *     mechanism, not a new one. The placeholder is marked
 *     XCE_WORKER_UNAVAIL (unjoined, dead-until-connected) and carries the
 *     restored admin bits. Its hostattrs_t is non-NULL with valid sentinel
 *     strings (copy_worker, wobj.c:53, requires non-NULL attrs; downstream
 *     apply_join_state()/add_hostattrs_resvals() treat hosttype/hostmodel
 *     as always-valid strings) so the host's later real join -- which
 *     join.c routes to rejoin_worker(), unmodified, because key_find() now
 *     finds this placeholder -- doesn't crash on rejoin_worker's direct
 *     attrs->hosttype dereference (worker.c:376-379).
 *
 * Idempotent under retry (5.8): find first via our own key_find, create
 * only if truly missing. A second call for the same host (e.g. a retried
 * restore after an earlier attempt failed partway through the snapshot)
 * finds either the real worker or the already-created placeholder and just
 * OR's the bits in again -- never a duplicate-create conflict.
 *
 * admin_status is masked to XCE_WORKER_ADMIN_MASK defensively (mirrors the
 * collect side's own discipline, collect_host_frames only ever encoding
 * admin-masked bits) so a caller passing an unmasked status word can never
 * leak a non-admin bit (e.g. XCE_WORKER_BUSY) onto the restored host.
 */
int restore_host_state(cluster_t* c, const char* name, uint32_t admin_status)
{
    worker_t* w  = NULL;
    int       rc = 0;

    if (name == NULL || name[0] == '\0')
        return -EINVAL;
    if (c == NULL || c->workermgr == NULL)
        return -EINVAL;

    admin_status &= XCE_WORKER_ADMIN_MASK;

    /* No reference() pin across the workermgr-unlock/w-lock gap (unlike
     * join.c's live-join path): this is called from checkpoint restore
     * (ha_handle_snapshot_chunk -> persist_load_snapshot), which DOES run
     * on the live vtask pool, not a quiesced phase, so a real first-time
     * join for the same host CAN race this call. That race is still safe:
     * a concurrent init_worker/makeWorker on a miss here would itself
     * find-or-create via the same key_find/makeWorker machinery under
     * workermgr's own lock, and either ordering converges to a worker
     * with the admin bits OR'd in and UNAVAIL cleared by the real join's
     * own apply_join_state() call -- the same "outer key_find, then
     * unconditionally call into makeWorker" shape already used elsewhere
     * in this codebase (worker_from_capnp/restore_worker,
     * create_lost_and_found). Mirrors restore_queue_state's (queue.c:208)
     * lock idiom, not a claim that this path is concurrency-free. */
    lock(c->workermgr, RTTI_LOCK_READ);
    rc = key_find(c->workermgr, (void*)name, (void**)&w);
    unlock(c->workermgr);

    if (rc < 0 && rc != -ENOENT)
        return rc;

    if (w != NULL) {
        /* Hit: real joined worker or an earlier placeholder -- only OR the
         * admin bits in, never touch anything else. */
        lock(w, RTTI_LOCK_WRITE);
        w->host.status |= admin_status;
        unlock(w);

        return 0;
    }

    /* Miss: create a minimal, valid placeholder -- mirrors the
     * lost_and_found sentinel host precedent (config_repl.c:1114-1130). */
    {
        hostattrs_t  placeholder_attrs = { .hosttype  = "unknown",
                                           .hostmodel = "unknown" };
        xce_worker_t placeholder_host  = { .name   = (char*)name,
                                           .attrs  = &placeholder_attrs,
                                           .status = XCE_WORKER_UNAVAIL
                                                     | admin_status };

        w = makeWorker(c->workermgr, &placeholder_host);
        if (w == NULL)
            return -errno;
    }

    return 0;
}

struct worker_config_publish_args {
    cluster_t*  c;
    endpoint_t* ep;
};

static int check_conf4worker_(xce_config_t* c)
{
    if (c != NULL) {
        return 1;
    }

    return 0;
}

static int worker_push_config(cluster_t* c, endpoint_t* ep)
{
    xce_configdata_t configdata;
    int              rc = 0;
    xlist_t          clist;

    memset(&configdata, 0x0, sizeof(configdata));

    configdata.data.conf4worker = calloc(1, sizeof(xce_conf4worker_t));
    if (configdata.data.conf4worker == NULL) {
        return -ENOMEM;
    }

    xlist_init(&clist, 256);

    collect_confvars(c, (collect_check_fn_t)check_conf4worker_, &clist);

    configdata.data.conf4worker->n_confvars = clist.count;
    configdata.data.conf4worker->confvars   = (xce_config_t**)clist.data;
    configdata.data_which                   = ConfigData_data_workerconfig;

    configdata.data.conf4worker->clustername
        = c->myconf->cluster_config->clustername;

    rc = xce_pushconfig_req(ep, c->myhost, &configdata);

    FREEUP(configdata.data.conf4worker);

    xlist_fini(&clist);

    return rc;
}

/* consider to push jobs on hosts to */
int on_worker_join(void* cls, endpoint_t* ep)
{
    cluster_t* c  = (cluster_t*)cls;
    int        rc = 0;

    rc = worker_push_config(c, ep);

    return rc;
}

int get_resval(cluster_t* c, char* resname, char* host, xce_val_t* val)
{
    xce_resource_t* r     = NULL;
    zarray_t*       array = NULL;
    xce_val_t*      v     = NULL;
    worker_t*       w     = NULL;
    int             rc    = 0;

    lock(c->resmgr, RTTI_LOCK_READ);
    rc = key_find(c->resmgr, (void*)resname, (void**)&r);
    unlock(c->resmgr);

    if (rc < 0) {
        return rc;
    }

    if (r == NULL) {
        return -ENOENT;
    }

    if ((r->flags & XCE_RESF_SHARED) == 0) {
        if ((host == NULL) || (strlen(host) == 0)) {

            return -EINVAL;
        }

        lock(c->workermgr, RTTI_LOCK_READ);
        rc = key_find(c->workermgr, (void*)host, (void**)&w);
        if (w != NULL) {
            reference(w);
        }
        unlock(c->workermgr);
        if (rc < 0) {
            return rc;
        }
        if (w == NULL) {
            return -ENOENT;
        }

        array = &(w->host.resvals);
    } else {
        array = &(c->sharedinst);
    }

    if (r->idx < array->caps) {
        v = (xce_val_t*)array->data[r->idx];
    }

    if (v == NULL) {
        if (w != NULL) {
            dereference(w);
        }

        return -ENOENT;
    }

    memcpy(val, v, sizeof(*val));

    if (w != NULL) {
        dereference(w);
    }

    return 1;
}

/* Local mutation body of "mark a worker dead" -- status flag, dead_since
 * (only if not already set), and the derived hostattrs/resvals reset. No
 * hook side effects and no locking of its own: callers hold w's lock (or,
 * for mark_worker_unavail, are about to fire the surrounding hooks) and
 * decide where dead_since comes from (wall clock live, or the WAL payload's
 * recorded timestamp on replay -- bug-1472-remainder). */
static void apply_worker_dead_mutation(cluster_t* c, worker_t* w,
                                       time_t dead_since)
{
    w->host.status |= XCE_WORKER_UNAVAIL;
    if (w->host.dead_since == 0)
        w->host.dead_since = dead_since;

    if (w->host.attrs != NULL) {
        reset_hostattrs_unknown(w->host.attrs);
        add_hostattrs_resvals(c, w);
        set_host_slots(c, w, 0);
    }
}

int mark_worker_unavail(cluster_t* c, worker_t* w)
{
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_WORKER_DEAD, c, w);

    apply_worker_dead_mutation(c, w, time(NULL));

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_WORKER_DEAD, c, w);
    return 0;
}

int mark_worker_back(cluster_t* c, worker_t* w)
{
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_WORKER_UPDATE, c, w);

    w->host.status &= ~XCE_WORKER_UNAVAIL;

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_WORKER_UPDATE, c, w);
    return 0;
}

int handle_worker_leave(cluster_t* c, int idx)
{
    worker_t* w = NULL;
    int       rc;

    TRACE("%s worker %d broken connection", __func__, idx);

    if (c == NULL) {
        SET_ERROR(EINVAL);
        return -errno;
    }

    if (idx < 0) {
        SET_ERROR(EINVAL);
        return -errno;
    }

    rc = vget(c->workermgr, idx, (void**)&w);
    if (rc < 0) {
        return rc;
    }

    if (w == NULL) {
        ERROR("fail to find worker by id %d", idx);
        SET_ERROR(E_XCE_BAD_WORKER);
        return -errno;
    }

    lock(w, RTTI_LOCK_WRITE);

    w->ep = NULL;

    mark_worker_unavail(c, w);

    unlock(w);

    /* bug-1693 (Decision Log: "No client reply-wait timeout as a fix for a
     * hung request -- fail loud at the real transport/source instead of
     * masking with a deadline backstop"). Wake any on_fetchfile() waiter
     * (job.c) proxied to this worker immediately instead of leaving it to
     * on_fetchfile's own 10s deadline -- mirrors handle_scheduler_leave's
     * identical schedadmin_signal_disconnect() fix in sched.c (same
     * Decision Log rule, same shared design mistake -- cluster.c's
     * OPCode_schedAdmin case comment already notes it "mirrors
     * on_fetchfile's proxy pattern"). Deliberately called AFTER releasing
     * w's lock: fetchfile_signal_worker_disconnect() only takes
     * g_fetchfile_mu (job.c, file-local) + each matched waiter's own mu,
     * never any worker_t lock, so there is no ordering dependency, and `w`
     * remains a valid pointer for the comparison (this function's vget'd
     * reference is still held). A no-op if no fetchfile request happens to
     * be pending against this worker. */
    fetchfile_signal_worker_disconnect(w);

    return 0;
}

int worker_ctrl(cluster_t* c, xce_hostctrl_t* ctrl)
{
    worker_t* w  = NULL;
    int       rc = 0;

    if (!c || !ctrl || !ctrl->hostname || *ctrl->hostname == '\0') {
        TRACE("%s invalid arguments.", __func__);
        return -EINVAL;
    }
    TRACE("%s worker_ctrl worker %s", __func__, ctrl->hostname);

    lock(c->workermgr, RTTI_LOCK_READ);
    rc = key_find(c->workermgr, (void*)ctrl->hostname, (void**)&w);
    if (w != NULL) {
        reference(w);
    }
    unlock(c->workermgr);

    if (rc < 0) {
        return rc;
    }

    if (w == NULL) {
        PERROR("workermgr_lock");
        return -ENOENT;
    }

    /* bug-447: key_find returns 1 on found (binary librtti, unnormalizable
     * upstream). Normalize the found-rc to 0 here so the AFTER_EVENT hook
     * gate below fires (WAL append + replication) and the handler's
     * commit-barrier pattern sees 0. Real errors are untouched: rc < 0
     * returned above, the w == NULL miss returned -ENOENT, and control-
     * action failures set rc themselves. */
    rc = 0;

    lock(w, RTTI_LOCK_WRITE);
    switch (ctrl->action) {
    case HOST_OPEN:
        if (!(w->host.status & XCE_WORKER_CLOSED)) {
            goto clean;
        }
        w->host.status &= ~(XCE_WORKER_CLOSED | XCE_WORKER_CLOSED_BY_ADMIN);
        break;
    case HOST_CLOSE:
        if (w->host.status & XCE_WORKER_CLOSED) {
            goto clean;
        }
        w->host.status |= (XCE_WORKER_CLOSED | XCE_WORKER_CLOSED_BY_ADMIN);
        break;
    default:
        rc = -1;
        break;
    }
    if (rc == 0) {
        xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER,
                      HOOK_AFTER_EVENT, HOOK_DETAIL_WORKER_UPDATE, c, w);
    }

clean:
    unlock(w);
    dereference(w);
    return rc;
}

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

    init_reply_by_req(&reply, request);

    rc = worker_ctrl(c, request->body.hostctrlmsg);

    /* Commit barrier (ha-commit-barrier-rollout Req 1.1-1.4): the success
     * reply is sent only after the operation's WAL entry is majority-
     * committed; barrier failures (-ETIMEDOUT, ...) become the error
     * reply. Errors from worker_ctrl reply immediately — no barrier wait.
     * worker_ctrl released all entity locks before returning, so blocking
     * here holds no lock. timeout 0 = configured sync_timeout_ms. */
    if (rc == 0) {
        rc = persist_commit_barrier(&c->persist, 0);
        if (rc != 0)
            WARN("%s commit barrier failed, rc %d", __func__, rc);
    }

    reply.status     = rc;
    reply.data_which = RequestReply_data_nodata;

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

/* Reconstruct a worker from a WAL_WORKER_CTRL Cap'n Proto payload.
 * Payload layout: [uint32_t detail][Cap'n Proto xce_worker_t].
 *
 * If the worker is already registered:
 *   - HOOK_DETAIL_WORKER_JOIN: a duplicate JOIN replay is genuinely a
 *     no-op -- returns -EEXIST, no mutation (unchanged, pre-bug-1472
 *     behavior).
 *   - HOOK_DETAIL_WORKER_UPDATE: the target already existing is the
 *     NORMAL case (worker_ctrl()'s HOST_OPEN/HOST_CLOSE, worker.c:732-794,
 *     only ever runs against an already-joined host). bug-1472: replace
 *     (not OR -- worker_ctrl uses assignment-style clear/set, so a
 *     HOST_OPEN must be able to CLEAR the closed bits) the existing
 *     worker's admin-masked status bits with the decoded payload's, then
 *     return 0 so the caller (persist.c's WAL_WORKER_CTRL apply case)
 *     falls through to fire the HOOK_DETAIL_WORKER_UPDATE notification
 *     instead of silently discarding the change.
 *   - HOOK_DETAIL_WORKER_DEAD (bug-1472-remainder): replays
 *     mark_worker_unavail()'s own local mutations on the existing worker --
 *     XCE_WORKER_UNAVAIL, dead_since (from the decoded payload's deadSince
 *     field, the PRIMARY's own death timestamp, not this member's clock),
 *     hostattrs reset to "unknown" sentinels, and the derived resvals
 *     re-computed from that reset state (add_hostattrs_resvals + slots=0
 *     via set_host_slots). This does NOT need a new wire field despite
 *     the original investigation's design-decision framing: the live
 *     xce_worker_t.resvals zarray_t the earlier analysis worried about
 *     never needs to travel on the wire at all, because
 *     reset-to-"unknown"-and-zero is a fixed, data-independent operation
 *     -- add_hostattrs_resvals/set_host_slots derive their values FROM
 *     w->host.attrs's (already-reset) fields, not from anything the
 *     primary computed. capnp_encode_worker (persist_worker_hook_fun)
 *     already serializes the FULL xce_worker_t.host on every WAL_WORKER_
 *     CTRL entry including DEAD, and mark_worker_unavail() calls
 *     reset_hostattrs_unknown() BEFORE firing the hook that triggers that
 *     encode -- so the Worker schema's existing `attrs` field was never
 *     actually the gap; this function's own DEAD branch simply never
 *     existed to apply anything. */
int worker_from_capnp(cluster_t* c, const void* capnp_payload, size_t len)
{
    if (!capnp_payload || len <= sizeof(uint32_t))
        return -EINVAL;

    uint32_t hook_detail;
    memcpy(&hook_detail, capnp_payload, sizeof(hook_detail));

    /* Skip the 4-byte detail discriminator written by persist_worker_hook_fun */
    const uint8_t* data     = (const uint8_t*)capnp_payload + sizeof(uint32_t);
    size_t         data_len = len - sizeof(uint32_t);

    struct capn   c_ctx;
    capn_init_mem(&c_ctx, data, data_len, 0);
    Worker_ptr wp;
    wp.p = capn_getp(capn_root(&c_ctx), 0, 1);
    struct Worker cw;
    xce_worker_t  host;
    memset(&host, 0, sizeof(host));
    read_Worker(&cw, wp);
    decode_Worker(&host, &cw);
    capn_free(&c_ctx);

    /* Idempotency guard: an existing worker means either a duplicate JOIN
     * replay (no-op) or an UPDATE targeting an already-joined host (the
     * normal case -- must be applied, see bug-1472). */
    worker_t* existing = NULL;
    lock(c->workermgr, RTTI_LOCK_READ);
    int found = key_find(c->workermgr, (void*)host.name, (void**)&existing);
    unlock(c->workermgr);
    if (found > 0 && existing != NULL) {
        if (hook_detail == HOOK_DETAIL_WORKER_UPDATE) {
            lock(existing, RTTI_LOCK_WRITE);
            existing->host.status
                = (existing->host.status & ~XCE_WORKER_ADMIN_MASK)
                | (host.status & XCE_WORKER_ADMIN_MASK);
            unlock(existing);

            free_Worker(&host);
            return 0;
        }

        if (hook_detail == HOOK_DETAIL_WORKER_DEAD) {
            lock(existing, RTTI_LOCK_WRITE);
            apply_worker_dead_mutation(c, existing, host.dead_since);
            unlock(existing);

            free_Worker(&host);
            return 0;
        }

        return -EEXIST;
    }

    /* bug-1427: distinguish "genuinely created a new worker" from a
     * negative errno, discarding restore_worker()'s raw idx (its own
     * return value, w->idx, is NOT a creation flag -- it can legitimately
     * be 0 for the first worker, which would collide with the
     * update-applied return above). Neither of this function's two
     * callers (persist.c) inspects the exact non-negative value, only
     * sign and -EEXIST, so this is a safe, backward-compatible
     * repurposing -- see apply_committed_entry()'s WAL_WORKER_CTRL case
     * and persist_verify_apply_correctness()'s new worker-count check. */
    int rc = restore_worker(c, host);
    return (rc < 0) ? rc : 1;
}

/* worker_should_retain lives in persist.c —
 * it only depends on xce_worker_t fields and persist_config_t,
 * and must be accessible from the unit test binary which links persist.c. */