myxce/src/frontend/apiserver/backend/clib.go

//go:build cgo

package backend

/*
#include <stdlib.h>
#include <stdint.h>
#include <errno.h>
#include "client.h"
#include "netio.h"
#include "helper.h"
#include "idgen.h"
#include "controller.capnp.h"
#include "val_helpers.h"
#include "xce/cluster.h"
#include "xce/errnox.h"

// backend_ctx_t wraps xcecli_ctx_t for use as the C handler's arg.
typedef struct {
    xcecli_ctx_t ctx;
} backend_ctx_t;

// go_push_handler is exported from Go and called from C for every incoming request.
extern int go_push_handler(xce_request_t*, void*);

static int xce_backend_handler(xce_request_t *req, void *arg) {
    return go_push_handler(req, arg);
}

// go_backend_rejoin_handler is exported from Go (reconnect.go) and
// registered below as the client library's xcecli_reconnect_cb
// (component-failover-adoption task 4.3). It only sets a flag and returns
// immediately -- see reconnect.go's own doc comment for why (the library
// invokes it while still holding ac->failover.lock).
extern int go_backend_rejoin_handler(xcecli_ctx_t, void*);

static int xce_backend_rejoin_handler(xcecli_ctx_t ctx, void *arg) {
    return go_backend_rejoin_handler(ctx, arg);
}

// xce_backend_init initialises the client context, sets the handler,
// starts the network loop, registers the rejoin callback, then joins as
// HostRole_apiserver.
static int xce_backend_init(xce_client_config_t *config, backend_ctx_t *ctx) {
    int rc = xcecli_ctx_init(&(ctx->ctx), config);
    if (rc < 0) return rc;
    xcecli_set_handler(ctx->ctx, (xce_request_handler_t)xce_backend_handler, NULL);
    rc = xcecli_start(ctx->ctx);
    if (rc < 0) return rc;
    // Registered AFTER xcecli_start() succeeds, not before
    // (component-failover-adoption task 4.3; matches worker's/scheduler's
    // task 2.3/3.1 fix): xcecli_start() unconditionally resets
    // ctx->failover.rejoin_cb to NULL as part of initializing per-context
    // failover state (client.c) -- registering before this call would be
    // silently wiped out immediately afterward.
    xcecli_set_reconnect_handler(ctx->ctx, xce_backend_rejoin_handler, NULL);
    return xcecli_join(ctx->ctx, HostRole_apiserver, NULL);
}

// Helpers to extract typed row pointers from a dataset_t.
static xce_job_t    **dataset_jobs(dataset_t *d)    { return (xce_job_t **)d->rows; }
static xce_worker_t **dataset_workers(dataset_t *d) { return (xce_worker_t **)d->rows; }

// CGO cannot access C union fields by name (they appear as [N]byte).
// These thin helpers set the correct union branch from C so Go callers
// don't need unsafe pointer arithmetic.
static void jobctrl_set_siginfo(xce_jobctrl_t *ctrl, xce_signal_t *sig) {
    ctrl->u.siginfo = sig;
}
static void jobctrl_set_ckptoption(xce_jobctrl_t *ctrl, xce_ckpt_options_t *ckpt) {
    ctrl->u.ckptoption = ckpt;
}

// xce_val_t union accessors moved to val_helpers.h (shared with convert.go).
*/
import "C"

import (
	"apiserver/models"
	logger "apiserver/utils"
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"unsafe"

	"go.uber.org/zap"
)

// gCtx holds the C client context for the lifetime of the process.
var gCtx C.backend_ctx_t

// gClusterCfg is the single validated xce_cluster_config_t RegisterToBackend
// builds on every call -- either by parsing config.Peers via the shared C
// parser (xce_parse_peers), or by synthesizing a one-candidate list from the
// legacy config.Address/config.Port fields when Peers is empty (Req 1 AC5: a
// single controller is the candidates_count=1 case of the identical
// validate/convert/connect pipeline, not a separately-maintained code path
// -- see buildClusterConfig() below).
//
// Package-level, not a buildClusterConfig()/RegisterToBackend()-local
// variable, for the same reason gCtx and gIDCtx already are (process-lifetime
// stable address): xce_cluster_config_to_client_config() (ConfigConverter,
// src/lib/cluster.c) aliases gClientCfg.masters[i] directly at this struct's
// own candidates[i].hostname fixed-char-array storage -- no strdup, no copy.
// A caller that built the source xce_cluster_config_t on the stack (or any
// Go-local variable) and then let gClientCfg outlive that scope would have a
// use-after-scope bug: confirmed by reading src/lib/client.c's
// xcecli_ctx_init() (`ctx->conf = cfg` -- stores a RAW pointer, never
// copies) and src/lib/client_failover.c (`ac->conf->masters[next]` is
// dereferenced on every failover rotation, not just at initial connect).
// Worker's and scheduler's C implementations hit this identical hazard
// (component-failover-adoption tasks 2.1/3.1) and solved it by keeping the
// source config on a process-lifetime struct (worker: cc_ctx->cluster_cfg;
// scheduler: sched->cluster_cfg) -- this package-level var is the direct Go
// equivalent.
var gClusterCfg C.xce_cluster_config_t

// gClientCfg is the xce_client_config_t ConfigConverter populates from
// gClusterCfg and xce_backend_init() connects with. Package-level for the
// same reason as gClusterCfg above: xcecli_ctx_init() stores gCtx.ctx->conf
// as a raw pointer to whichever struct address is passed to it, dereferenced
// for the ctx's entire runtime life (every failover rotation reads
// ctx->conf->masters[]/nmasters, not only the initial connect) -- so this
// struct's address must be stable for the process lifetime, not a
// RegisterToBackend-local variable that goes out of scope when the function
// returns.
var gClientCfg C.xce_client_config_t

// gIDCtx is the C idgen context used to generate job IDs.
var gIDCtx C.idgen_ctx_t
var idCtxOnce sync.Once

// gHandler receives push notifications from the backend.
// Must be set via StartNotificationListener before RegisterToBackend.
var gHandler NotificationHandler

// writeCCharArray copies s into dst, a fixed-size C char array field of
// gClusterCfg (hostname/identity/key), truncating to len(dst)-1 bytes and
// always leaving the result NUL-terminated. Because these are fixed arrays
// embedded directly in gClusterCfg (a package-level, process-lifetime
// struct -- see its doc comment), populating them this way needs no
// C.CString/C.free bookkeeping at all: xce_cluster_config_to_client_config()
// later aliases pointers directly into this same storage.
func writeCCharArray(dst []C.char, s string) {
	n := len(s)
	if n > len(dst)-1 {
		n = len(dst) - 1
	}
	for i := 0; i < n; i++ {
		dst[i] = C.char(s[i])
	}
	dst[n] = 0
}

// formatCandidateList renders gClusterCfg's current candidates as
// "host:port, host:port, ..." for diagnostic logging (Req 5.1, 5.2) -- the
// same textual shape scheduler's sched_format_candidate_list() (scheduler.c)
// produces, for cross-component log consistency.
func formatCandidateList() string {
	n := int(gClusterCfg.candidates_count)
	parts := make([]string, 0, n)
	for i := 0; i < n; i++ {
		cand := &gClusterCfg.candidates[i]
		parts = append(parts, fmt.Sprintf("%s:%d", C.GoString(&cand.hostname[0]), int(cand.port)))
	}
	return strings.Join(parts, ", ")
}

// currentClusterCandidates returns gClusterCfg's current candidates as a
// Go-native slice. Exists so clib_test.go (a //go:build cgo test file that
// cannot `import "C"` itself -- a Go-toolchain restriction, golang/go#13938)
// can assert on buildClusterConfig()'s output shape without re-implementing
// CGO field access in the test file, mirroring the candidates.go/
// candidates_test.go split already established in this package.
func currentClusterCandidates() []Candidate {
	n := int(gClusterCfg.candidates_count)
	out := make([]Candidate, n)
	for i := 0; i < n; i++ {
		cand := &gClusterCfg.candidates[i]
		out[i] = Candidate{
			Name: C.GoString(&cand.name[0]),
			Host: C.GoString(&cand.hostname[0]),
			Port: int(cand.port),
		}
	}
	return out
}

// currentClientConfigNMasters exposes gClientCfg.nmasters as a plain int,
// for the same test-file-cannot-import-C reason as currentClusterCandidates
// above. Lets tests confirm xce_cluster_config_to_client_config() actually
// propagated the full candidate count into the struct xce_backend_init()
// connects with -- proving ONE call carries all N candidates, rather than N
// separate single-candidate calls (the deleted per-peer Go loop's behavior,
// which always hardcoded nmasters=1).
func currentClientConfigNMasters() int {
	return int(gClientCfg.nmasters)
}

// buildClusterConfig builds and validates the single xce_cluster_config_t
// this connect attempt will use (Req 1 AC1-3/AC5), writing directly into the
// package-level gClusterCfg (see its doc comment for why this must be
// package-level rather than a local variable).
//
// A single controller is not a separately-maintained code path: when
// config.Peers is non-empty it is parsed via the same shared C parser
// (xce_parse_peers) worker and scheduler already call directly; otherwise a
// one-candidate list is synthesized from config.Address/config.Port. Either
// source is then run through the identical xce_cluster_config_validate()
// call below -- no branch downstream of this function distinguishes the two
// sources (Req 1 AC5), mirroring scheduler's sched_build_cluster_config()
// (scheduler.c) and worker's equivalent (controller_connector.c).
//
// This intentionally does not reuse parseCandidates() (candidates.go):
// parseCandidates() parses into its own function-local xce_cluster_config_t
// and reconstructs a Go []Candidate slice from it -- exactly the
// use-after-scope-prone, throwaway-C-struct pattern gClusterCfg's doc
// comment above warns against for this call site. buildClusterConfig() must
// parse directly into the package-level gClusterCfg so the very same struct
// flows unmodified into xce_cluster_config_to_client_config() below.
//
// Does not call xce_cluster_config_to_client_config() or connect -- callers
// (RegisterToBackend) do that separately.
func buildClusterConfig(config *BackendConfig) error {
	from := config.From
	if from == "" {
		from = "xce_apiserver"
	}
	key := config.Key
	if key == "" {
		key = "keepsecret$"
	}

	gClusterCfg = C.xce_cluster_config_t{}

	if len(config.Peers) > 0 {
		peersStr := strings.Join(config.Peers, ",")
		cPeers := C.CString(peersStr)
		rc := C.xce_parse_peers(cPeers, &gClusterCfg)
		C.free(unsafe.Pointer(cPeers))
		if rc != 0 {
			return fmt.Errorf(
				"apiserver: --peers parse error: expected \"[name=]host:port,...\" got %q",
				peersStr)
		}
	} else {
		gClusterCfg.candidates_count = 1
		writeCCharArray(gClusterCfg.candidates[0].hostname[:], config.Address)
		gClusterCfg.candidates[0].port = C.int(config.Port)
	}

	writeCCharArray(gClusterCfg.identity[:], from)
	writeCCharArray(gClusterCfg.key[:], key)

	if rc := C.xce_cluster_config_validate(&gClusterCfg); rc != 0 {
		return fmt.Errorf(
			"apiserver: invalid controller candidate configuration (rc=%d): %s",
			int(rc), formatCandidateList())
	}

	return nil
}

// RegisterToBackend initialises the XCE client library and joins as an API
// server, using the full configured candidate list -- config.Peers if set,
// otherwise a single-candidate list synthesized from config.Address/
// config.Port (Req 1 AC5: the candidates_count=1 case of the identical
// pipeline, not a separate code path -- see buildClusterConfig()). Builds
// exactly ONE validated xce_cluster_config_t regardless of source, converts
// it once via the shared ConfigConverter
// (xce_cluster_config_to_client_config), and calls xce_backend_init()
// exactly ONCE -- replacing the former dual-mechanism implementation (a
// Go-level per-peer tryConnect() retry loop for the Peers case, and a
// second, separately-coded tryConnect() call for the legacy Address/Port
// case; both bypassed xce_cluster_config_t/xce_cluster_config_validate()/
// xce_cluster_config_to_client_config() entirely, and both always
// hardcoded nmasters=1).
//
// StartNotificationListener should be called first so that push events have
// a handler ready before the first push can arrive.
func RegisterToBackend(config *BackendConfig) error {
	if err := buildClusterConfig(config); err != nil {
		return err
	}

	logger.Info("Attempting backend connection",
		zap.Int("candidates", int(gClusterCfg.candidates_count)),
		zap.String("candidate_list", formatCandidateList()))

	gClientCfg = C.xce_client_config_t{}
	if rc := C.xce_cluster_config_to_client_config(&gClusterCfg, &gClientCfg); rc != 0 {
		return fmt.Errorf("apiserver: candidate list conversion failed (rc=%d)", int(rc))
	}
	// ConfigConverter deliberately does not set from/key (caller-owned
	// identity/auth, per its own doc comment) -- point them at gClusterCfg's
	// own identity/key fixed arrays rather than allocating separate C
	// strings: gClusterCfg is the same package-level, process-lifetime
	// struct ConfigConverter already aliased gClientCfg.masters[i] into, so
	// this needs no additional C.CString/C.free bookkeeping.
	gClientCfg.from = &gClusterCfg.identity[0]
	gClientCfg.key = &gClusterCfg.key[0]

	rc := C.xce_backend_init(&gClientCfg, &gCtx)
	if rc < 0 {
		logger.Warn("All configured backend candidates failed",
			zap.Int("candidates", int(gClusterCfg.candidates_count)),
			zap.String("candidate_list", formatCandidateList()),
			zap.Int("rc", int(rc)))
		return fmt.Errorf("xce_backend_init failed across %d candidate(s) (rc=%d): %s",
			int(gClusterCfg.candidates_count), int(rc), formatCandidateList())
	}

	logger.Info("Backend connection established",
		zap.Int("nmasters", int(gClientCfg.nmasters)),
		zap.String("candidate_list", formatCandidateList()))
	return nil
}

// StartNotificationListener registers the handler that receives backend push
// events.  Call this before RegisterToBackend.
func StartNotificationListener(handler NotificationHandler) error {
	gHandler = handler
	return nil
}

// initIDGen initialises the C idgen context. Called once via sync.Once.
func initIDGen() {
	C.idgen_ctx_init(&gIDCtx)
}

// GenerateJobID calls the C idgen() to produce a unique ordinary job ID.
// clusterID and apiserverID identify the originating instance. arraySize
// must be 1: idgen() allocates ordinary IDs only and returns -EINVAL for
// any other size; array IDs come from GenerateArrayJobID.
func GenerateJobID(clusterID, apiserverID int, arraySize int) (uint64, error) {
	idCtxOnce.Do(initIDGen)
	var jobID C.uint64_t
	rc := C.idgen(&gIDCtx, C.size_t(clusterID), C.size_t(apiserverID),
		C.size_t(arraySize), &jobID)
	if rc < 0 {
		return 0, fmt.Errorf("idgen failed: %d", int(rc))
	}
	return uint64(jobID), nil
}

// GenerateArrayJobID calls the C idgen_array() to allocate a new array of n
// elements (1 <= n <= 32767, numbered from 1) and returns the first element's
// ID (tag TAG_JOB_ARRAY, index 1); ExtractArrayBaseID of it is the array ID.
// n == 1 is a one-element array, not an ordinary job (job-array-one-based-index
// 1.1, 1.2). Errors on a bad n or when the sequence block would overflow.
func GenerateArrayJobID(clusterID, apiserverID int, n int) (uint64, error) {
	if n < 1 {
		return 0, fmt.Errorf("idgen_array: invalid array size %d", n)
	}
	idCtxOnce.Do(initIDGen)
	var first C.uint64_t
	rc := C.idgen_array(&gIDCtx, C.size_t(clusterID), C.size_t(apiserverID),
		C.size_t(n), &first)
	if rc < 0 {
		return 0, fmt.Errorf("idgen_array failed: %d", int(rc))
	}
	return uint64(first), nil
}

// Cleanup releases C resources. Call at process shutdown.
func Cleanup() {
	C.idgen_ctx_fini(&gIDCtx)
}

// currentReconnectGen reads endpoint_t.reconnect_gen (netio.h) off the
// backend's live endpoint via the existing, previously-unused C accessor
// xcecli_get_my_endpoint() -- no new C getter needed (component-failover-
// adoption Phase 6, bug-1548). Uses an explicit atomic load (rather than a
// bare field dereference) so the Go-side read is formally atomic and pairs
// correctly with the C11 _Atomic write in xce_ctx_reconnect() (xcectx.c).
func currentReconnectGen() uint64 {
	ep := C.xcecli_get_my_endpoint(gCtx.ctx)
	if ep == nil {
		return 0
	}
	return atomic.LoadUint64((*uint64)(unsafe.Pointer(&ep.reconnect_gen)))
}

// StartHeartbeat launches a goroutine that sends periodic xcecli_heartbeat
// keep-alive messages to the controller. Stops when ctx is cancelled.
//
// Wires runHeartbeat's reconnect-detected and every-successful-tick hooks to
// reconnect.go's setNeedsRejoin/checkAndPerformRejoin (task 4.3) -- closing
// the idle-period reconnect gap (task 4.4/Phase 6, Req 2 AC4/AC5):
// XCECLI_FF_CALL (backing xcecli_heartbeat) never invokes the client
// library's rejoin callback, so nothing else would set the pending-rejoin
// flag if this heartbeat prober is the only traffic during a controller
// outage. currentReconnectGen (above) is what actually detects the
// reconnect even when it completed silently inside a single XCECLI_FF_CALL.
func StartHeartbeat(ctx context.Context) {
	go runHeartbeat(ctx, func(secs int) int {
		return int(C.xcecli_heartbeat(gCtx.ctx, C.int(secs)))
	}, currentReconnectGen, setNeedsRejoin, checkAndPerformRejoin)
}

// go_push_handler is called by the C library on every incoming request.
// It must be exported so the C trampoline can call it from a C thread.
//
//export go_push_handler
func go_push_handler(req *C.xce_request_t, _ unsafe.Pointer) C.int {
	if req == nil {
		return 0
	}
	defer C.free_Request(req)

	if gHandler == nil {
		return 0
	}

	switch req.opcode {
	case C.OPCode_push:
		return handlePush(req)
	case C.OPCode_notify:
		if req.body_which != C.Request_body_notifymsg {
			return 0
		}
		nd := (*C.xce_notify_data_t)(unsafe.Pointer(C.xce_request_body(req)))
		if nd == nil {
			return 0
		}
		return handleNotifyData(nd)
	}
	return 0
}

// handlePush dispatches an OPCode_push request to the appropriate handler.
func handlePush(req *C.xce_request_t) C.int {
	switch req.body_which {
	case C.Request_body_datamsg:
		dataset := (*C.dataset_t)(unsafe.Pointer(C.xce_request_body(req)))
		return handleDataset(dataset)
	case C.Request_body_configpushmsg:
		cfgdata := (*C.xce_configdata_t)(unsafe.Pointer(C.xce_request_body(req)))
		return handleConfigPush(cfgdata)
	}
	return 0
}

// handleDataset processes a dataset push (jobs or workers).
func handleDataset(d *C.dataset_t) C.int {
	switch d.table {
	case C.XceData_data_jobs:
		n := int(d.nrows)
		if n == 0 {
			return 0
		}
		jobs := unsafe.Slice(C.dataset_jobs(d), n)
		for _, jp := range jobs {
			if jp == nil {
				continue
			}
			info := cJobToModel(jp)
			gHandler.OnJobUpdated(info)
		}

	case C.XceData_data_workers:
		n := int(d.nrows)
		if n == 0 {
			return 0
		}
		workers := unsafe.Slice(C.dataset_workers(d), n)
		for _, wp := range workers {
			if wp == nil {
				continue
			}
			info := cWorkerToModel(wp)
			gHandler.OnHostUpdated(info)
		}

	case C.XceData_data_jobArray:
		// job-array-single-wal-record task 5.4 (Req 6.2, 6.4, 6.5, 8.4):
		// sibling receive path to the XceData_data_jobs case above, for the
		// compact array descriptor apiserver_push_jobarray() (apiserver.c,
		// task 5.2) sends. Unlike the LIST-shaped cases, extract_data()
		// (request.c, task 5.3b) always delivers exactly one descriptor
		// (dataset->nrows == 1, dataset->rows == the single
		// *xce_job_submit_t) -- no per-row slice to walk. Decode it,
		// expand locally into N models.JobInfo entries, and hand them to
		// dispatchJobArray for dedup + delivery + confirmation logging.
		if d.nrows == 0 || d.rows == nil {
			return 0
		}
		jobArray := (*C.xce_job_submit_t)(d.rows)
		elements := cJobArrayToModels(jobArray)
		dispatchJobArray(gHandler, elements, uint64(jobArray.jobid), uint64(jobArray.array_size))
	}
	return 0
}

// handleConfigPush processes a cluster configuration push.
func handleConfigPush(cfgdata *C.xce_configdata_t) C.int {
	if gHandler == nil {
		return 0
	}
	if cfgdata.data_which != C.ConfigData_data_clusterconfig {
		return 0
	}

	conf := (*C.cluster_config_t)(C.xce_configdata_ptr(cfgdata))
	if conf == nil {
		return 0
	}

	cluster := cClusterConfToClusterInfo(conf)
	if cluster == nil {
		logger.Warn("handleConfigPush: cluster_config_t has empty clustername; skipping OnClusterUpdated")
	}
	dispatchConfigPush(gHandler,
		cClusterConfQueues(conf),
		cClusterConfResources(conf),
		cClusterConfHostConfigs(conf),
		cluster,
	)
	gHandler.OnUserGroupsUpdated(cClusterConfUserGroups(conf))
	return 0
}

// handleNotifyData processes an OPCode_notify request.
// xce_notify_data_t uses 'event' (not 'data_which') as its union discriminant.
func handleNotifyData(nd *C.xce_notify_data_t) C.int {
	if nd == nil {
		return 0
	}
	switch nd.event {
	case C.NotifyData_data_updateJob:
		jp := (*C.xce_job_t)(C.xce_notify_data_ptr(nd))
		if jp == nil {
			return 0
		}
		gHandler.OnJobUpdated(cJobToModel(jp))

	case C.NotifyData_data_jobFinishInfo:
		fi := (*C.xce_jobfinish_t)(C.xce_notify_data_ptr(nd))
		if fi == nil {
			return 0
		}
		gHandler.OnJobFinished(uint64(fi.jobid), int(fi.exit_code))

	case C.NotifyData_data_jobFailInfo:
		ff := (*C.job_failinfo_t)(C.xce_notify_data_ptr(nd))
		if ff == nil {
			return 0
		}
		gHandler.OnJobFinished(uint64(ff.jobid), int(ff.code))

	case C.NotifyData_data_schedUpdateResvals:
		rvl := (*C.xce_resval_list_t)(C.xce_notify_data_ptr(nd))
		if rvl == nil {
			return 0
		}
		n := int(rvl.count)
		if n == 0 || rvl.vals == nil {
			return 0
		}
		vals := unsafe.Slice(rvl.vals, n)
		for _, vp := range vals {
			if vp == nil {
				continue
			}
			rd := cResvalToDistribution(vp)
			if rd != nil {
				gHandler.OnResourceLocationUpdated(rd)
			}
		}
	}
	return 0
}

// ErrBackendUnavailable is returned immediately by callBackend (and
// therefore by every wrapper below) when ApiserverBackendHealth reports the
// backend connection degraded (Req 4.1) -- a clear, distinct error a caller
// can recognize as "backend down, retrying" rather than an unrelated
// internal failure, without ever invoking the underlying blocking
// C.xcecli_* call.
var ErrBackendUnavailable = errors.New(
	"apiserver: backend connection unavailable (circuit breaker open); try again shortly")

// backendDegradedTTL is how long callBackend marks the backend degraded
// (ApiserverBackendHealth.MarkDegraded) after a call's own outcome
// indicates its retry budget was exhausted, before request handlers resume
// probing it directly again -- the half-open half of the circuit breaker
// (design.md's ApiserverBackendHealth). Chosen well below xcecli's own
// full per-call retry budget (XCECLI_FAILOVER_CALL: max_retries=20,
// backoff capped at 5000ms -- roughly 1-2 minutes total,
// client_failover.c:218,220): 10s is short enough that a backend which
// recovers quickly gets re-probed promptly, while still long enough that
// every request arriving during a genuine outage fails fast instead of
// each independently re-discovering the same ~100s exhausted budget one
// blocking call at a time.
const backendDegradedTTL = 10 * time.Second

// notReadyRC exposes -E_XCE_NOT_READY as a plain Go int, for
// health_rejoin_test.go (a //go:build cgo test file that cannot `import "C"`
// itself, golang/go#13938) to script a callBackend closure returning this
// exact retry-budget-exhaustion value without hardcoding the raw errno
// number -- same plain-Go-signature test-seam rationale as performRejoinFn
// (reconnect.go) and currentClientConfigNMasters() (above).
var notReadyRC = -int(C.E_XCE_NOT_READY)

// callBackend is the single shared pre-call/post-call helper every
// gCtx.ctx-based backend-call wrapper below (SubmitJob, ControlJob,
// ControlQueue, ControlHost, FetchFile) calls around its own single
// C.xcecli_* invocation (component-failover-adoption task 4.5, boundary:
// ApiserverBackendBridge/ApiserverBackendHealth) -- implemented once here,
// not copy-pasted per wrapper. fn performs exactly one C.xcecli_* call and
// returns its raw rc as a plain Go int (not C.int) -- the same
// plain-Go-signature test seam pattern as reconnect.go's performRejoinFn,
// letting a //go:build cgo test file (which can never `import "C"` itself,
// golang/go#13938) drive this helper directly with a scripted closure, with
// no live backend connection or even an initialized gCtx required.
//
// Pre-call (Req 4.1): if IsDegraded(), fn is never invoked at all -- the
// wrapped call (and its own up-to-~100s xcecli retry budget) is skipped
// entirely, returning ErrBackendUnavailable immediately. checkAndPerformRejoin
// is deliberately NOT called on this path either: it can itself block on
// xcecli_join's own up-to-~100s retry budget, which would defeat the whole
// point of failing fast here (Req 4.1's bounded, sub-second response). The
// rejoin-retry cadence during this window keeps running regardless, driven
// by heartbeat.go's own unconditional per-tick call to checkAndPerformRejoin
// (task 4.4) -- which never reads IsDegraded() -- so the pending-rejoin
// flag's retry-until-actually-joined behavior is never suppressed by this
// short-circuit (see health.go's own doc comment for why these two states
// must stay independent).
//
// Post-call (only reached when not degraded): the existing rejoin check
// (checkAndPerformRejoin, task 4.3/4.4, unchanged) always runs, regardless
// of fn's own outcome -- a pending rejoin is orthogonal to whether this
// particular call happened to succeed. If fn's rc is exactly
// -E_XCE_NOT_READY (this call's own retry budget exhausted), MarkDegraded
// opens the circuit for subsequent calls.
func callBackend(name string, fn func() int) (rc int, err error) {
	if IsDegraded() {
		return 0, ErrBackendUnavailable
	}

	rc = fn()
	checkAndPerformRejoin()

	if rc == notReadyRC {
		MarkDegraded(backendDegradedTTL)
	}
	if rc < 0 {
		return rc, fmt.Errorf("%s failed: %d", name, rc)
	}
	return rc, nil
}

// SubmitJob sends a job submission to the backend via xcecli_jobsubmit.
// arraySize=1 for normal jobs, >1 for array jobs.
// The store is updated only when the backend pushes the resulting OnJobCreated
// notification — not immediately here.
func SubmitJob(job *models.JobInfo, arraySize int) error {
	js, sd, si := modelJobToSubmit(job)
	defer freeJobSubmit(js, sd, si)

	js.array_size = C.uint64_t(arraySize)

	var outJobID C.uint64_t
	_, err := callBackend("xcecli_jobsubmit", func() int {
		return int(C.xcecli_jobsubmit(gCtx.ctx, js, &outJobID))
	})
	if err != nil {
		return err
	}
	if outJobID != 0 {
		job.Unique_JobID = int64(outJobID)
		job.JobID = int64(ExtractArrayBaseID(uint64(outJobID)))
		job.ArrayIdx = ExtractArrayIdx(uint64(outJobID))
	}
	return nil
}

// ControlJob sends a job control action to the backend via xcecli_jobctrl.
// action must be one of the models.JOB_* constants; content is the
// JSON-encoded models.JobControlEvent (signal, force_kill, method) or a bare
// signal name/number -- see jobCtrlParams (convert_pure.go).
func ControlJob(jobid uint64, action int, content string) error {
	var ctrl C.xce_jobctrl_t
	ctrl.jobid = C.uint64_t(jobid)
	// bug-1781/gitea#45: xce_jobctrl_t.action holds the JobControl_u_which
	// capnp union discriminant, not the job_ctrl_type_t action code -- the
	// two enums must not be conflated.
	ctrl.action = C.int(jobCtrlUnionDiscriminant(action))

	// For signal-based actions set signo; for checkpoint set ckpt options.
	// xce_jobctrl_t.u is a C union — we use a static C helper to set the
	// right branch safely.
	args, err := jobCtrlParams(action, content)
	if err != nil {
		return err
	}
	switch action {
	case models.JOB_STOP, models.JOB_RESUME, models.JOB_KILL:
		var sig C.xce_signal_t
		sig.signo = C.int(args.Signo)
		sig.force = C.int(args.Force)
		C.jobctrl_set_siginfo(&ctrl, &sig)
	case models.JOB_CHKPNT:
		var ckpt C.xce_ckpt_options_t
		ckpt.kill = C.int(args.CkptKill)
		if args.Method != "" {
			// NULL makes xce_jobctrl_req() send "default"
			ckpt.method = C.CString(args.Method)
			defer C.free(unsafe.Pointer(ckpt.method))
		}
		C.jobctrl_set_ckptoption(&ctrl, &ckpt)
	}

	_, err = callBackend("xcecli_jobctrl", func() int {
		return int(C.xcecli_jobctrl(gCtx.ctx, &ctrl))
	})
	return err
}

// ControlQueue sends a queue open/close/activate/deactivate action to the
// backend via xcecli_queuectrl.
func ControlQueue(queuename string, action int) error {
	cQueue := C.CString(queuename)
	defer C.free(unsafe.Pointer(cQueue))

	ctrl := C.xce_queuectrl_t{
		action: C.int(action),
		queue:  cQueue,
	}

	_, err := callBackend("xcecli_queuectrl", func() int {
		return int(C.xcecli_queuectrl(gCtx.ctx, &ctrl))
	})
	return err
}

// ControlHost sends a host open/close action to the backend via xcecli_hostctrl.
func ControlHost(hostname string, action int) error {
	cHost := C.CString(hostname)
	defer C.free(unsafe.Pointer(cHost))

	ctrl := C.xce_hostctrl_t{
		action:   C.int(action),
		hostname: cHost,
	}

	_, err := callBackend("xcecli_hostctrl", func() int {
		return int(C.xcecli_hostctrl(gCtx.ctx, &ctrl))
	})
	return err
}

// FetchFile fetches up to maxBytes of a job output file from offset via the
// controller→worker chain. The filepath must be the fully resolved absolute
// path on the exec host.
func FetchFile(jobid uint64, filepath string, offset uint64, maxBytes uint32) (*FetchFileResult, error) {
	cFilepath := C.CString(filepath)
	defer C.free(unsafe.Pointer(cFilepath))

	msg := C.xce_fetchfile_t{
		jobid:     C.uint64_t(jobid),
		filepath:  cFilepath,
		offset:    C.uint64_t(offset),
		max_bytes: C.uint32_t(maxBytes),
	}

	var reply *C.xce_fetchfile_reply_t
	_, err := callBackend("xcecli_fetchfile", func() int {
		return int(C.xcecli_fetchfile(gCtx.ctx, &msg, &reply))
	})
	if err != nil {
		return nil, err
	}
	if reply == nil {
		return nil, fmt.Errorf("xcecli_fetchfile returned nil reply")
	}
	defer C.free(unsafe.Pointer(reply))

	if reply.error != 0 {
		return nil, fmt.Errorf("fetchfile remote error: %d", int(reply.error))
	}

	result := &FetchFileResult{
		FileSize:  uint64(reply.file_size),
		BytesRead: uint32(reply.bytes_read),
	}
	if reply.bytes_read > 0 && reply.content != nil {
		result.Content = C.GoBytes(unsafe.Pointer(reply.content), C.int(reply.bytes_read))
	}
	return result, nil
}