wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity
myxce/docs/superpowers/plans/2026-05-07-p25-discovery.md
Rongsong Shen 59597e105e feat: add --peers to controller for HA peer list and priority election 
- myconf_t gains peers_cfg (xce_cluster_config_t) populated from --peers
- controller/main.c: parse --peers host:port,... via argp; ca_cert_path
  auto-derived from workdir/certs/xce.crt (no --ca-cert flag needed)
- cluster_init: derive my_priority from position in peers list (index 0 =
  primary); fall back to 0 when --peers not given (single-node mode)
2026-05-08 04:40:27 +00:00

1247 lines
40 KiB
Markdown
Raw Blame History

# P2.5: libxce Smart Discovery — TDD Implementation Plan
**Branch**: `persistent_ha`
**Depends on**: P1 complete
**Date**: 2026-05-07
---
## Overview
P2.5 adds client-side controller discovery and auto-reconnect to libxce, then plumbs it into worker, scheduler, and the Go API server. After P2.5, no component connects to the controller via a hardcoded single address — all use the candidate list and SSL.
P2.5-7 (CGo bridge in clib.go) is **deferred** — noted at end of document as future work.
P2.5-4 is **removed** — no `xce_load_cluster_config()`. Callers populate `xce_cluster_config_t` directly from CLI flags.
---
## Task Index
| Task | File(s) | Description |
|------|---------|-------------|
| P2.5-1 | `src/lib/include/xce/cluster.h` (new) | Define `xce_candidate_t`, `xce_cluster_config_t`, function prototypes |
| P2.5-2 | `src/lib/cluster.c` (new) | `xce_connect_cluster()`: SSL connect with candidate fallback + redirect follow |
| P2.5-2a | `src/lib/meson.build` | Add `cluster.c` to `libxce_src` |
| P2.5-3 | `src/lib/cluster.c` | `xce_reconnect()` auto-reconnect state machine |
| P2.5-5 | `src/worker/src/main.c` | Parse `--peers` / `--ca-cert`; use `xce_cluster_config_t` |
| P2.5-6 | `src/scheduler/src/main.c` | Same pattern as worker |
| P2.5-7a | `src/frontend/apiserver/models/jobs.go` + `jobs_test.go` | Map `JOB_STAT_TRANSIENT` (0x400) → `"TRANSIENT"` in `JobStateStr` |
| P2.5-8 | `src/lib/testsuite/test_cluster.c` + meson | Candidate selection, redirect, backoff, reconnect state machine |
---
## P2.5-1: `src/lib/include/xce/cluster.h`
### Files to create
- `src/lib/include/xce/cluster.h` (new directory + file — `include/xce/` does not yet exist)
### Note on SSL return type
`xce_connect_cluster()` is specified to return `SSL*`. However, `endpoint_t` in `netio.h` already wraps both `fd` and `ssl` fields together, and `iomgr_t` owns the `SSL_CTX*`. The implementation in cluster.c will **return a newly-allocated `endpoint_t*`** (which contains the SSL* inside), matching the actual usage pattern in `client.c`. This is the practical interpretation — "SSL connection" means an initialised endpoint with the TLS handshake complete. Document this deviation in the header with a `/* returns endpoint_t* — SSL* is ep->ssl */` comment.
### RED — failing test
The test itself is in P2.5-8; this task only defines the header. The build fails until `cluster.h` exists because `test_cluster.c` includes it.
### GREEN — create the header
```c
/* src/lib/include/xce/cluster.h */
#ifndef XCE_CLUSTER_H
#define XCE_CLUSTER_H
#include <stddef.h>
#include "config.h" /* MAX_HA_CANDIDATES, XCE_MAXHOSTNAME_LEN */
/**
* A single controller candidate (host + port).
*/
typedef struct {
char hostname[XCE_MAXHOSTNAME_LEN];
int port;
} xce_candidate_t;
/**
* Cluster connection configuration.
* Populated by each caller (worker/scheduler/apiserver) from --peers and
* --ca-cert CLI flags. libxce never reads config files.
*
* candidates_count must be <= MAX_HA_CANDIDATES (4).
* A WARN is logged at startup if candidates_count > 3 (validated for <= 3).
*/
typedef struct {
xce_candidate_t candidates[MAX_HA_CANDIDATES];
int candidates_count;
char ca_cert_path[4096]; /* path to CA certificate for TLS verify */
char identity[256]; /* "from" name sent to controller */
char key[256]; /* authentication key */
} xce_cluster_config_t;
/**
* Reconnect state machine states.
*/
typedef enum {
XCE_RECONNECT_IDLE = 0,
XCE_RECONNECT_CONNECTING,
XCE_RECONNECT_BACKOFF,
XCE_RECONNECT_CONNECTED,
XCE_RECONNECT_FAILED
} xce_reconnect_state_t;
/**
* Reconnect context (opaque to callers — use xce_reconnect_*).
*/
typedef struct {
xce_cluster_config_t* config;
xce_reconnect_state_t state;
int attempt;
int backoff_ms; /* current backoff in ms */
void* endpoint; /* endpoint_t* when connected */
} xce_reconnect_ctx_t;
/* --- API --- */
/**
* xce_cluster_config_validate() — validate a populated xce_cluster_config_t.
*
* Returns 0 on success, -EINVAL if:
* - candidates_count < 1 or > MAX_HA_CANDIDATES
* - any candidate has empty hostname or port <= 0
* - ca_cert_path is empty
* - identity is empty
* Logs WARN if candidates_count > 3.
*/
int xce_cluster_config_validate(const xce_cluster_config_t* cfg);
/**
* xce_connect_cluster() — connect to the cluster via candidate list.
*
* Iterates candidates in order, attempting a TLS connection to each.
* If a candidate responds with a REDIRECT reply, follows the redirect
* (up to XCE_MAX_REDIRECTS times) to the actual primary.
*
* Returns: endpoint_t* (cast to void*) on success, NULL on failure.
* The SSL* is accessible as ((endpoint_t*)result)->ssl.
*
* The caller is responsible for calling endpoint_fini() when done.
*
* Note: This function performs a synchronous blocking connect + TLS
* handshake on each candidate. It does NOT start the iomgr event loop.
* After a successful return, the caller must register the endpoint
* with an iomgr_t before reading/writing requests.
*/
void* xce_connect_cluster(xce_cluster_config_t* cfg);
/**
* xce_reconnect_init() — initialise a reconnect context.
*/
int xce_reconnect_init(xce_reconnect_ctx_t* rctx, xce_cluster_config_t* cfg);
/**
* xce_reconnect_step() — drive one step of the reconnect state machine.
*
* Call in a loop until state == XCE_RECONNECT_CONNECTED or
* XCE_RECONNECT_FAILED. The function may sleep up to backoff_ms before
* returning when in XCE_RECONNECT_BACKOFF state.
*
* Returns 0 to continue, -1 when permanently failed.
*/
int xce_reconnect_step(xce_reconnect_ctx_t* rctx);
/**
* xce_reconnect_fini() — release resources held by a reconnect context.
*/
void xce_reconnect_fini(xce_reconnect_ctx_t* rctx);
/* Redirect follow limit */
#define XCE_MAX_REDIRECTS 4
/* Backoff parameters */
#define XCE_RECONNECT_INITIAL_BACKOFF_MS 500
#define XCE_RECONNECT_MAX_BACKOFF_MS 30000
#define XCE_RECONNECT_BACKOFF_MULTIPLIER 2
#endif /* XCE_CLUSTER_H */
```
### Commit
```bash
git add src/lib/include/xce/cluster.h
git commit -m "feat: add xce_cluster_config_t and cluster discovery API (P2.5-1)"
```
---
## P2.5-2 + P2.5-2a: `src/lib/cluster.c` and `meson.build`
### Files to create/modify
- `src/lib/cluster.c` (new)
- `src/lib/meson.build` (add `'cluster.c'` to `libxce_src`)
### Implementation notes
- **SSL**: Use `SSL_CTX_set_verify(ctx, SSL_VERIFY_PEER | SSL_VERIFY_FAIL_IF_NO_PEER_CERT, NULL)` on the client SSL context. The TLS library rejects invalid certificates before application logic runs.
- **Redirect protocol**: After a successful TLS connect, send a `Hello` request. If the response type is `Response_redirect`, parse `redirect_host` and `redirect_port` from the capnp message and reconnect to that address (up to `XCE_MAX_REDIRECTS`). If the response is `Response_ok`, the connection is live.
- **Candidate fallback**: If a candidate is unreachable (connect/handshake fails), move to the next candidate with no delay. Backoff is only applied between full candidate list sweeps.
- **Stub note**: The unit tests (P2.5-8) mock the actual socket/SSL operations via a function pointer hook (`_xce_connect_raw_fn`) so tests do not require real SSL. The production code path calls the real `socket_connect` + `SSL_connect` sequence inherited from `netio.c` patterns.
### GREEN — minimal cluster.c
```c
/* src/lib/cluster.c */
#include <errno.h>
#include <stddef.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include "xce/cluster.h"
#include "netio.h"
#include "utils.h"
/* Test seam: override in unit tests to avoid real TCP/SSL. */
typedef void* (*connect_raw_fn_t)(const char* host, int port,
const char* ca_cert);
static connect_raw_fn_t _connect_raw_fn = NULL;
void xce_cluster_set_connect_fn(connect_raw_fn_t fn)
{
_connect_raw_fn = fn;
}
/* Forward decl for the real connect (lives at bottom of file). */
static void* cluster_connect_real(const char* host, int port,
const char* ca_cert);
static void* do_connect(const char* host, int port, const char* ca_cert)
{
if (_connect_raw_fn != NULL) {
return _connect_raw_fn(host, port, ca_cert);
}
return cluster_connect_real(host, port, ca_cert);
}
int xce_cluster_config_validate(const xce_cluster_config_t* cfg)
{
if (cfg == NULL) {
return -EINVAL;
}
if (cfg->candidates_count < 1 || cfg->candidates_count > MAX_HA_CANDIDATES) {
return -EINVAL;
}
for (int i = 0; i < cfg->candidates_count; i++) {
if (cfg->candidates[i].hostname[0] == '\0') {
return -EINVAL;
}
if (cfg->candidates[i].port <= 0) {
return -EINVAL;
}
}
if (cfg->ca_cert_path[0] == '\0') {
return -EINVAL;
}
if (cfg->identity[0] == '\0') {
return -EINVAL;
}
if (cfg->candidates_count > 3) {
fprintf(stderr,
"WARN: HA cluster has %d candidates; LSF-style election is "
"validated for <=3. Larger clusters increase split-brain risk.\n",
cfg->candidates_count);
}
return 0;
}
void* xce_connect_cluster(xce_cluster_config_t* cfg)
{
if (xce_cluster_config_validate(cfg) != 0) {
return NULL;
}
for (int sweep = 0; sweep < 2; sweep++) {
for (int i = 0; i < cfg->candidates_count; i++) {
const char* host = cfg->candidates[i].hostname;
int port = cfg->candidates[i].port;
/* Follow redirects from this candidate */
char redir_host[XCE_MAXHOSTNAME_LEN];
int redir_port = port;
strncpy(redir_host, host, sizeof(redir_host) - 1);
redir_host[sizeof(redir_host) - 1] = '\0';
for (int r = 0; r <= XCE_MAX_REDIRECTS; r++) {
void* ep = do_connect(redir_host, redir_port, cfg->ca_cert_path);
if (ep == NULL) {
break; /* try next candidate */
}
/* Check for redirect reply (stubbed in tests via connect fn) */
/* In production: send Hello request, parse Response.
* If redirect, update redir_host/redir_port and loop.
* If ok, return ep. */
/*
* NOTE: Full capnp Hello/Response round-trip requires the
* iomgr event loop. In the initial implementation we treat a
* successful TLS connect as "connected" and let the caller's
* first real request detect a redirect response and call
* xce_connect_cluster() again. The test seam below lets unit
* tests simulate redirect behaviour without a real controller.
*/
return ep;
}
}
}
return NULL;
}
int xce_reconnect_init(xce_reconnect_ctx_t* rctx, xce_cluster_config_t* cfg)
{
if (rctx == NULL || cfg == NULL) {
return -EINVAL;
}
if (xce_cluster_config_validate(cfg) != 0) {
return -EINVAL;
}
rctx->config = cfg;
rctx->state = XCE_RECONNECT_IDLE;
rctx->attempt = 0;
rctx->backoff_ms = XCE_RECONNECT_INITIAL_BACKOFF_MS;
rctx->endpoint = NULL;
return 0;
}
int xce_reconnect_step(xce_reconnect_ctx_t* rctx)
{
if (rctx == NULL) {
return -1;
}
switch (rctx->state) {
case XCE_RECONNECT_IDLE:
rctx->state = XCE_RECONNECT_CONNECTING;
/* fall through */
case XCE_RECONNECT_CONNECTING: {
void* ep = xce_connect_cluster(rctx->config);
if (ep != NULL) {
rctx->endpoint = ep;
rctx->state = XCE_RECONNECT_CONNECTED;
rctx->attempt = 0;
rctx->backoff_ms = XCE_RECONNECT_INITIAL_BACKOFF_MS;
return 0;
}
/* Connection failed — enter backoff */
rctx->attempt++;
rctx->state = XCE_RECONNECT_BACKOFF;
return 0;
}
case XCE_RECONNECT_BACKOFF: {
int sleep_ms = rctx->backoff_ms;
if (sleep_ms > XCE_RECONNECT_MAX_BACKOFF_MS) {
sleep_ms = XCE_RECONNECT_MAX_BACKOFF_MS;
}
/* Add ±25% jitter */
int jitter = (sleep_ms / 4) * ((rctx->attempt % 3) - 1);
sleep_ms += jitter;
if (sleep_ms < 0) {
sleep_ms = XCE_RECONNECT_INITIAL_BACKOFF_MS;
}
struct timespec ts = { .tv_sec = sleep_ms / 1000,
.tv_nsec = (sleep_ms % 1000) * 1000000L };
nanosleep(&ts, NULL);
/* Advance backoff for next failure */
rctx->backoff_ms *= XCE_RECONNECT_BACKOFF_MULTIPLIER;
rctx->state = XCE_RECONNECT_CONNECTING;
return 0;
}
case XCE_RECONNECT_CONNECTED:
return 0;
case XCE_RECONNECT_FAILED:
return -1;
default:
return -1;
}
}
void xce_reconnect_fini(xce_reconnect_ctx_t* rctx)
{
if (rctx == NULL) {
return;
}
/* endpoint_t cleanup is caller's responsibility — they registered it with
* iomgr and must call endpoint_fini() themselves. */
rctx->endpoint = NULL;
rctx->state = XCE_RECONNECT_IDLE;
}
/* --- Real SSL connect (production path) --- */
static void* cluster_connect_real(const char* host, int port,
const char* ca_cert)
{
/*
* Full implementation requires:
* 1. socket_create() + socket_connect() from netio.c internals
* 2. SSL_CTX_new(TLS_client_method())
* 3. SSL_CTX_load_verify_locations(ctx, ca_cert, NULL)
* 4. SSL_CTX_set_verify(ctx, SSL_VERIFY_PEER |
* SSL_VERIFY_FAIL_IF_NO_PEER_CERT, NULL)
* 5. endpoint_init() + endpoint set fd/ssl
* 6. SSL_connect() + handshake completion
*
* These are internal netio.c helpers not yet exposed in netio.h.
* Tracked as future work: expose endpoint_connect_ssl() from netio.c
* and call it here. Until then, callers in worker/scheduler use the
* existing xcecli_ctx_init() + endpoint_connect() path (which already
* handles SSL via the iomgr) and pass a single candidate.
*/
(void)host;
(void)port;
(void)ca_cert;
return NULL; /* not yet implemented — use test seam in unit tests */
}
```
### meson.build edit
Add `'cluster.c'` to `libxce_src` list, after `'client.c'`:
```
# In libxce_src, after 'client.c':
'cluster.c',
```
Also add `test_cluster.c` to `libxce_test_src` (done in P2.5-8).
### Commit
```bash
git add src/lib/cluster.c src/lib/meson.build
git commit -m "feat: implement xce_connect_cluster and cluster.c skeleton (P2.5-2, P2.5-2a)"
```
---
## P2.5-3: `xce_reconnect()` auto-reconnect state machine
The state machine is already implemented in P2.5-2 as `xce_reconnect_init/step/fini`. This task is the **integration confirmation** step — verifying that a caller can drive the state machine to `XCE_RECONNECT_CONNECTED` and that the backoff advances correctly. The tests are in P2.5-8.
No additional source changes are needed if P2.5-2 is correct. If P2.5-8 tests reveal gaps, fix cluster.c here.
### Commit (if no changes needed)
```bash
# No separate commit — P2.5-3 is satisfied by P2.5-2 implementation + P2.5-8 tests passing.
```
---
## P2.5-5: Update `src/worker/src/main.c`
### Files to modify
- `src/worker/src/main.c`
### Context
The worker currently parses `-m remote_host` and `-p remote_port` into `config.remote_host` / `config.remote_port` (lines 395, 403 of `main.c`). It then calls `endpoint_connect()` with a single address. After P2.5-5, it instead parses `--peers` and `--ca-cert` into an `xce_cluster_config_t` and calls `xce_connect_cluster()`.
### RED — failing test
There are no unit tests for `main.c` directly (it is an integration boundary). The failing condition is: after the change, the build must succeed and the integration test (P3-1) must pass. For TDD purposes, write a **helper function test** for the peer-list parser before implementing it.
Add `src/worker/src/parse_peers.c` + `src/worker/src/parse_peers.h` with a pure function `parse_peers_arg()` that can be unit tested without the full main():
```c
/* src/worker/src/parse_peers_test.c — compile with unity */
#include <string.h>
#include "parse_peers.h"
#include "xce/cluster.h"
#include <unity/unity.h>
static void test_parse_single_peer(void)
{
xce_cluster_config_t cfg;
memset(&cfg, 0, sizeof(cfg));
int rc = parse_peers_arg("host1:9990", &cfg);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(1, cfg.candidates_count);
TEST_ASSERT_EQUAL_STRING("host1", cfg.candidates[0].hostname);
TEST_ASSERT_EQUAL_INT(9990, cfg.candidates[0].port);
}
static void test_parse_two_peers(void)
{
xce_cluster_config_t cfg;
memset(&cfg, 0, sizeof(cfg));
int rc = parse_peers_arg("host1:9990,host2:9991", &cfg);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(2, cfg.candidates_count);
TEST_ASSERT_EQUAL_STRING("host2", cfg.candidates[1].hostname);
TEST_ASSERT_EQUAL_INT(9991, cfg.candidates[1].port);
}
static void test_parse_too_many_peers_returns_error(void)
{
xce_cluster_config_t cfg;
memset(&cfg, 0, sizeof(cfg));
/* MAX_HA_CANDIDATES = 4; 5 entries must be rejected */
int rc = parse_peers_arg("h1:1,h2:2,h3:3,h4:4,h5:5", &cfg);
TEST_ASSERT_NOT_EQUAL(0, rc);
}
static void test_parse_missing_port_returns_error(void)
{
xce_cluster_config_t cfg;
memset(&cfg, 0, sizeof(cfg));
int rc = parse_peers_arg("host1", &cfg); /* no colon:port */
TEST_ASSERT_NOT_EQUAL(0, rc);
}
static void test_parse_empty_string_returns_error(void)
{
xce_cluster_config_t cfg;
memset(&cfg, 0, sizeof(cfg));
int rc = parse_peers_arg("", &cfg);
TEST_ASSERT_NOT_EQUAL(0, rc);
}
void test_parse_peers(void)
{
RUN_TEST(test_parse_single_peer);
RUN_TEST(test_parse_two_peers);
RUN_TEST(test_parse_too_many_peers_returns_error);
RUN_TEST(test_parse_missing_port_returns_error);
RUN_TEST(test_parse_empty_string_returns_error);
}
```
### GREEN — parse_peers.h
```c
/* src/worker/src/parse_peers.h */
#ifndef XCE_PARSE_PEERS_H
#define XCE_PARSE_PEERS_H
#include "xce/cluster.h"
/**
* parse_peers_arg() — parse a comma-separated "host:port,host:port" string
* into cfg->candidates[].
*
* Returns 0 on success, -EINVAL on malformed input or too many candidates.
* Sets cfg->candidates_count.
*/
int parse_peers_arg(const char* peers_str, xce_cluster_config_t* cfg);
#endif
```
### GREEN — parse_peers.c
```c
/* src/worker/src/parse_peers.c */
#include "parse_peers.h"
#include "config.h"
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
int parse_peers_arg(const char* peers_str, xce_cluster_config_t* cfg)
{
if (peers_str == NULL || peers_str[0] == '\0' || cfg == NULL) {
return -EINVAL;
}
/* Work on a mutable copy */
char buf[4096];
strncpy(buf, peers_str, sizeof(buf) - 1);
buf[sizeof(buf) - 1] = '\0';
int count = 0;
char* saveptr;
char* token = strtok_r(buf, ",", &saveptr);
while (token != NULL) {
if (count >= MAX_HA_CANDIDATES) {
return -EINVAL; /* too many candidates */
}
char* colon = strrchr(token, ':');
if (colon == NULL) {
return -EINVAL; /* missing port */
}
*colon = '\0';
char* host = token;
int port = atoi(colon + 1);
if (host[0] == '\0' || port <= 0) {
return -EINVAL;
}
strncpy(cfg->candidates[count].hostname, host,
XCE_MAXHOSTNAME_LEN - 1);
cfg->candidates[count].hostname[XCE_MAXHOSTNAME_LEN - 1] = '\0';
cfg->candidates[count].port = port;
count++;
token = strtok_r(NULL, ",", &saveptr);
}
if (count == 0) {
return -EINVAL;
}
cfg->candidates_count = count;
return 0;
}
```
### GREEN — main.c changes
Replace the existing `-m` / `-p` option parsing in `main.c` with `--peers` and `--ca-cert`:
```c
/* In options[] argp table — replace existing -m/-p with: */
{ "peers", 'P', "HOST:PORT[,HOST:PORT]", 0,
"Comma-separated controller candidate list (max 4)" },
{ "ca-cert", 'C', "FILE", 0,
"Path to CA certificate for controller TLS verification" },
/* In parse_opt(): */
case 'P':
if (parse_peers_arg(arg, &cluster_cfg) != 0) {
argp_error(state, "invalid --peers value: %s", arg);
}
break;
case 'C':
strncpy(cluster_cfg.ca_cert_path, arg,
sizeof(cluster_cfg.ca_cert_path) - 1);
break;
/* In main(), after argp_parse():
* Populate identity/key from existing config path, then:
*/
strncpy(cluster_cfg.identity, config.virutal_name, /* existing field */
sizeof(cluster_cfg.identity) - 1);
if (xce_cluster_config_validate(&cluster_cfg) != 0) {
fprintf(stderr, "ERROR: invalid cluster configuration\n");
return 1;
}
/* Replace existing endpoint_connect() call with: */
void* ep = xce_connect_cluster(&cluster_cfg);
if (ep == NULL) {
fprintf(stderr, "ERROR: failed to connect to any controller candidate\n");
return 1;
}
/* ... use ep (cast to endpoint_t*) where config.remote_host/port was used */
```
**Backward compatibility**: Keep `-m` / `-p` as deprecated aliases that populate a single-entry `xce_cluster_config_t`. Log a WARN when deprecated flags are used.
### Commit
```bash
git add src/worker/src/parse_peers.c src/worker/src/parse_peers.h \
src/worker/src/parse_peers_test.c src/worker/src/main.c
git commit -m "feat: worker uses --peers/--ca-cert via xce_cluster_config_t (P2.5-5)"
```
---
## P2.5-6: Update `src/scheduler/src/main.c`
### Files to modify
- `src/scheduler/src/main.c`
- `src/scheduler/src/parse_peers.c` (copy/symlink from worker, or share via lib)
### Context
The scheduler currently hardcodes `config->masters[0] = "localhost"` and `config->master_ports[0] = 9990` in `init_config()`. After P2.5-6, it parses `--peers` and `--ca-cert` from CLI into `xce_cluster_config_t`.
The `parse_peers_arg()` function is identical to worker's version. Options:
1. Copy `parse_peers.c` + `parse_peers.h` into `src/scheduler/src/` (simplest, YAGNI — no shared lib abstraction yet).
2. Add a small shared `src/lib/cluster_cli.c` file later if a third caller appears.
**Decision**: Copy into scheduler (option 1). Refactor to shared lib when a third caller appears.
### RED — test (in scheduler testsuite or reuse worker test binary)
Same `test_parse_peers` tests apply — run the same parser tests against the scheduler's copy. Add to scheduler's meson test target.
### GREEN — scheduler main.c changes
Pattern is identical to worker (P2.5-5):
```c
/* Replace init_config() hardcoded values with CLI parsing */
static struct argp_option options[] = {
{ "logfile", 'l', "FILE", 0, "Full path of log file" },
{ "verbose", 'v', 0, 0, "Verbose log" },
{ "peers", 'P', "HOST:PORT[,HOST:PORT]", 0,
"Controller candidate list (max 4)" },
{ "ca-cert", 'C', "FILE", 0, "CA certificate path" },
{ 0 }
};
/* parse_opt() handles 'P' and 'C' as in worker */
/* In main(): validate cluster_cfg, then call xce_connect_cluster()
* instead of xcecli_ctx_init() with a single master. */
```
### Commit
```bash
git add src/scheduler/src/parse_peers.c src/scheduler/src/parse_peers.h \
src/scheduler/src/main.c
git commit -m "feat: scheduler uses --peers/--ca-cert via xce_cluster_config_t (P2.5-6)"
```
---
## P2.5-7a: `JOB_STAT_TRANSIENT` serialization in Go API server
### Files to modify
- `src/frontend/apiserver/models/jobs.go`
- `src/frontend/apiserver/models/jobs_test.go`
### Context
`JOB_STAT_TRANSIENT = 0x400` is defined in `src/include/xce/types.h`.
`JOBSTAT` map in `models/jobs.go` does not yet contain `"TRANSIENT"`.
`JobStateStr()` function falls through to `return "ERR"` for bitmask 0x400.
The fix has two parts:
1. Add `"TRANSIENT": 0x400` to the `JOBSTAT` map.
2. Add a check for `JOB_STAT_TRANSIENT` in `JobStateStr()` **before** the existing `PEND` check, so a job with only the TRANSIENT bit set returns `"TRANSIENT"` immediately.
### RED — failing test (add to `jobs_test.go`)
```go
// In TestJobStateStr test cases, add:
{
name: "TRANSIENT (0x400 — state uncertain after controller failover)",
state: JOBSTAT["TRANSIENT"],
want: "TRANSIENT",
},
{
name: "TRANSIENT with RUN bit (RUN job marked uncertain at promotion)",
state: JOBSTAT["RUN"] | JOBSTAT["TRANSIENT"],
want: "TRANSIENT",
},
```
Run `go test ./models/...` — these cases currently return `"ERR"` (TRANSIENT) and `"RUN"` (RUN|TRANSIENT) respectively. Both fail.
### GREEN — jobs.go changes
```go
// In var JOBSTAT map, add after "UNKNOWN":
"TRANSIENT": 0x400, // JOB_STAT_TRANSIENT: state uncertain after controller failover
// In JobStateStr(), add as the FIRST check before the PEND check:
func JobStateStr(state int) string {
if (state & JOBSTAT["TRANSIENT"]) != 0 {
return "TRANSIENT"
}
if (state & JOBSTAT["PEND"]) != 0 {
// ... existing logic unchanged
```
The TRANSIENT check must come first because a job can have `JOB_STAT_RUN | JOB_STAT_TRANSIENT` set simultaneously (was running, now uncertain). The TRANSIENT state supersedes the last-known state for display purposes — showing "RUN" would actively mislead users.
### Additional test: status_detail
The design doc specifies a `status_detail` field for TRANSIENT jobs. This is surfaced at the API response layer (controller → HTTP response), not in `JobStateStr()`. Add a separate test for the response builder:
```go
// In convert_test.go or a new models/transient_test.go:
func TestTransientJobResponseDetail(t *testing.T) {
// When a job has TRANSIENT bit set, the response must include
// status_detail: "job state uncertain after controller failover — awaiting worker reconnect"
// This test is a placeholder — implement when the response struct gains status_detail field.
// Mark as TODO pending P2.5-7 (CGo bridge) which adds the field to JobInfo.
t.Skip("status_detail field not yet in JobInfo struct — tracked in P2.5-7")
}
```
### Commit
```bash
git add src/frontend/apiserver/models/jobs.go \
src/frontend/apiserver/models/jobs_test.go
git commit -m "feat: map JOB_STAT_TRANSIENT (0x400) to TRANSIENT in JobStateStr (P2.5-7a)"
```
---
## P2.5-8: Tests for discovery, reconnect, redirect, backoff
### Files to create/modify
- `src/lib/testsuite/test_cluster.c` (new)
- `src/lib/meson.build` — add `test_cluster.c` to `libxce_test_src`
- `src/lib/testsuite/test.c` — add `extern void test_cluster(void)` and `test_cluster()` call
### Test seam API additions to cluster.c/cluster.h
Add to the **private** header (only used from tests — not installed):
```c
/* src/lib/cluster_testseam.h — not installed, included only by test_cluster.c */
#ifndef XCE_CLUSTER_TESTSEAM_H
#define XCE_CLUSTER_TESTSEAM_H
#include "xce/cluster.h"
typedef void* (*connect_raw_fn_t)(const char* host, int port,
const char* ca_cert);
/** Override the raw connect function for unit tests. Pass NULL to restore. */
void xce_cluster_set_connect_fn(connect_raw_fn_t fn);
#endif
```
### RED — test_cluster.c (write before cluster.c passes these)
```c
/* src/lib/testsuite/test_cluster.c */
#include <errno.h>
#include <stdio.h>
#include <string.h>
#include <unity.h>
#include "xce/cluster.h"
#include "cluster_testseam.h"
/* --- test seam state --- */
static int s_connect_call_count = 0;
static int s_connect_fail_first_n = 0; /* fail first N calls */
static int s_redirect_on_call = -1; /* -1 = never redirect */
static char s_redirect_host[256];
static int s_redirect_port = 0;
static int s_calls_before_success = 0;
/* Minimal fake endpoint type for tests */
typedef struct { int fake_fd; } fake_ep_t;
static fake_ep_t s_fake_ep;
static void* mock_connect(const char* host, int port, const char* ca_cert)
{
(void)host;
(void)port;
(void)ca_cert;
s_connect_call_count++;
if (s_connect_fail_first_n > 0) {
s_connect_fail_first_n--;
return NULL;
}
/* Redirect simulation: caller (xce_connect_cluster) must re-call with
* redirect target. In this simplified stub, we track that the right
* host/port are passed when redirect is expected. */
s_fake_ep.fake_fd = 42;
return &s_fake_ep;
}
static void reset_seam(void)
{
s_connect_call_count = 0;
s_connect_fail_first_n = 0;
s_redirect_on_call = -1;
s_redirect_host[0] = '\0';
s_redirect_port = 0;
xce_cluster_set_connect_fn(mock_connect);
}
/* --- helper: build a valid config --- */
static void make_cfg(xce_cluster_config_t* cfg,
const char* peers[], const int ports[], int n)
{
memset(cfg, 0, sizeof(*cfg));
for (int i = 0; i < n; i++) {
strncpy(cfg->candidates[i].hostname, peers[i],
XCE_MAXHOSTNAME_LEN - 1);
cfg->candidates[i].port = ports[i];
}
cfg->candidates_count = n;
strncpy(cfg->ca_cert_path, "/tmp/test-ca.pem",
sizeof(cfg->ca_cert_path) - 1);
strncpy(cfg->identity, "test-worker", sizeof(cfg->identity) - 1);
}
/* ============================================================
* Validation tests
* ============================================================ */
static void test_validate_rejects_null(void)
{
int rc = xce_cluster_config_validate(NULL);
TEST_ASSERT_NOT_EQUAL(0, rc);
}
static void test_validate_rejects_zero_candidates(void)
{
xce_cluster_config_t cfg;
memset(&cfg, 0, sizeof(cfg));
cfg.candidates_count = 0;
strncpy(cfg.ca_cert_path, "/ca.pem", sizeof(cfg.ca_cert_path) - 1);
strncpy(cfg.identity, "x", sizeof(cfg.identity) - 1);
TEST_ASSERT_NOT_EQUAL(0, xce_cluster_config_validate(&cfg));
}
static void test_validate_rejects_too_many_candidates(void)
{
xce_cluster_config_t cfg;
memset(&cfg, 0, sizeof(cfg));
cfg.candidates_count = MAX_HA_CANDIDATES + 1;
TEST_ASSERT_NOT_EQUAL(0, xce_cluster_config_validate(&cfg));
}
static void test_validate_rejects_empty_ca_cert(void)
{
const char* h[] = { "host1" };
const int p[] = { 9990 };
xce_cluster_config_t cfg;
make_cfg(&cfg, h, p, 1);
cfg.ca_cert_path[0] = '\0';
TEST_ASSERT_NOT_EQUAL(0, xce_cluster_config_validate(&cfg));
}
static void test_validate_rejects_empty_identity(void)
{
const char* h[] = { "host1" };
const int p[] = { 9990 };
xce_cluster_config_t cfg;
make_cfg(&cfg, h, p, 1);
cfg.identity[0] = '\0';
TEST_ASSERT_NOT_EQUAL(0, xce_cluster_config_validate(&cfg));
}
static void test_validate_rejects_bad_port(void)
{
xce_cluster_config_t cfg;
memset(&cfg, 0, sizeof(cfg));
strncpy(cfg.candidates[0].hostname, "host1", XCE_MAXHOSTNAME_LEN - 1);
cfg.candidates[0].port = 0; /* invalid */
cfg.candidates_count = 1;
strncpy(cfg.ca_cert_path, "/ca.pem", sizeof(cfg.ca_cert_path) - 1);
strncpy(cfg.identity, "x", sizeof(cfg.identity) - 1);
TEST_ASSERT_NOT_EQUAL(0, xce_cluster_config_validate(&cfg));
}
static void test_validate_accepts_valid_single(void)
{
const char* h[] = { "ctrl1" };
const int p[] = { 9990 };
xce_cluster_config_t cfg;
make_cfg(&cfg, h, p, 1);
TEST_ASSERT_EQUAL_INT(0, xce_cluster_config_validate(&cfg));
}
static void test_validate_accepts_max_candidates(void)
{
const char* h[] = { "h1", "h2", "h3", "h4" };
const int p[] = { 1, 2, 3, 4 };
xce_cluster_config_t cfg;
make_cfg(&cfg, h, p, MAX_HA_CANDIDATES);
TEST_ASSERT_EQUAL_INT(0, xce_cluster_config_validate(&cfg));
}
/* ============================================================
* Candidate selection tests
* ============================================================ */
static void test_connect_cluster_returns_non_null_on_success(void)
{
reset_seam();
const char* h[] = { "ctrl1" };
const int p[] = { 9990 };
xce_cluster_config_t cfg;
make_cfg(&cfg, h, p, 1);
void* ep = xce_connect_cluster(&cfg);
TEST_ASSERT_NOT_NULL(ep);
TEST_ASSERT_EQUAL_INT(1, s_connect_call_count);
}
static void test_connect_cluster_tries_second_on_first_fail(void)
{
reset_seam();
s_connect_fail_first_n = 1; /* first candidate fails */
const char* h[] = { "ctrl1", "ctrl2" };
const int p[] = { 9990, 9990 };
xce_cluster_config_t cfg;
make_cfg(&cfg, h, p, 2);
void* ep = xce_connect_cluster(&cfg);
TEST_ASSERT_NOT_NULL(ep);
/* must have tried ctrl1 (fail) then ctrl2 (success) */
TEST_ASSERT_EQUAL_INT(2, s_connect_call_count);
}
static void test_connect_cluster_returns_null_if_all_fail(void)
{
reset_seam();
s_connect_fail_first_n = 999; /* all fail */
const char* h[] = { "ctrl1", "ctrl2" };
const int p[] = { 9990, 9991 };
xce_cluster_config_t cfg;
make_cfg(&cfg, h, p, 2);
void* ep = xce_connect_cluster(&cfg);
TEST_ASSERT_NULL(ep);
}
static void test_connect_cluster_rejects_invalid_config(void)
{
reset_seam();
xce_cluster_config_t cfg;
memset(&cfg, 0, sizeof(cfg));
/* candidates_count = 0 — invalid */
void* ep = xce_connect_cluster(&cfg);
TEST_ASSERT_NULL(ep);
/* mock must NOT have been called — validation rejects before connect */
TEST_ASSERT_EQUAL_INT(0, s_connect_call_count);
}
/* ============================================================
* Reconnect state machine tests
* ============================================================ */
static void test_reconnect_init_sets_idle_state(void)
{
const char* h[] = { "ctrl1" };
const int p[] = { 9990 };
xce_cluster_config_t cfg;
make_cfg(&cfg, h, p, 1);
xce_reconnect_ctx_t rctx;
int rc = xce_reconnect_init(&rctx, &cfg);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(XCE_RECONNECT_IDLE, (int)rctx.state);
TEST_ASSERT_EQUAL_INT(0, rctx.attempt);
TEST_ASSERT_EQUAL_INT(XCE_RECONNECT_INITIAL_BACKOFF_MS, rctx.backoff_ms);
TEST_ASSERT_NULL(rctx.endpoint);
}
static void test_reconnect_init_rejects_null_config(void)
{
xce_reconnect_ctx_t rctx;
int rc = xce_reconnect_init(&rctx, NULL);
TEST_ASSERT_NOT_EQUAL(0, rc);
}
static void test_reconnect_step_transitions_to_connected(void)
{
reset_seam();
/* mock always succeeds */
const char* h[] = { "ctrl1" };
const int p[] = { 9990 };
xce_cluster_config_t cfg;
make_cfg(&cfg, h, p, 1);
xce_reconnect_ctx_t rctx;
xce_reconnect_init(&rctx, &cfg);
/* Step once — IDLE -> CONNECTING -> CONNECTED */
int rc = xce_reconnect_step(&rctx);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(XCE_RECONNECT_CONNECTED, (int)rctx.state);
TEST_ASSERT_NOT_NULL(rctx.endpoint);
xce_reconnect_fini(&rctx);
}
static void test_reconnect_step_enters_backoff_on_fail(void)
{
reset_seam();
s_connect_fail_first_n = 999; /* always fail */
const char* h[] = { "ctrl1" };
const int p[] = { 9990 };
xce_cluster_config_t cfg;
make_cfg(&cfg, h, p, 1);
xce_reconnect_ctx_t rctx;
xce_reconnect_init(&rctx, &cfg);
/* Step 1: IDLE -> CONNECTING -> fails -> BACKOFF */
xce_reconnect_step(&rctx);
/* State is either BACKOFF (if step stays async) or CONNECTING after
* sleeping; either way attempt must be > 0 */
TEST_ASSERT_TRUE(rctx.attempt > 0 ||
rctx.state == XCE_RECONNECT_BACKOFF);
}
static void test_reconnect_backoff_advances(void)
{
reset_seam();
s_connect_fail_first_n = 999;
const char* h[] = { "ctrl1" };
const int p[] = { 9990 };
xce_cluster_config_t cfg;
make_cfg(&cfg, h, p, 1);
xce_reconnect_ctx_t rctx;
xce_reconnect_init(&rctx, &cfg);
int initial_backoff = rctx.backoff_ms;
/* Two failure steps */
xce_reconnect_step(&rctx); /* IDLE->CONNECTING->fail->BACKOFF */
xce_reconnect_step(&rctx); /* BACKOFF->sleep->CONNECTING->fail->BACKOFF */
/* backoff must have doubled (or be capped at max) */
TEST_ASSERT_TRUE(rctx.backoff_ms >= initial_backoff * XCE_RECONNECT_BACKOFF_MULTIPLIER
|| rctx.backoff_ms == XCE_RECONNECT_MAX_BACKOFF_MS);
}
static void test_reconnect_resets_backoff_on_connect(void)
{
reset_seam();
s_connect_fail_first_n = 2; /* fail twice, then succeed */
const char* h[] = { "ctrl1" };
const int p[] = { 9990 };
xce_cluster_config_t cfg;
make_cfg(&cfg, h, p, 1);
xce_reconnect_ctx_t rctx;
xce_reconnect_init(&rctx, &cfg);
/* Drive until connected (with real nanosleep — test intentionally slow) */
int iters = 0;
while (rctx.state != XCE_RECONNECT_CONNECTED && iters < 20) {
xce_reconnect_step(&rctx);
iters++;
}
TEST_ASSERT_EQUAL_INT(XCE_RECONNECT_CONNECTED, (int)rctx.state);
TEST_ASSERT_EQUAL_INT(0, rctx.attempt);
TEST_ASSERT_EQUAL_INT(XCE_RECONNECT_INITIAL_BACKOFF_MS, rctx.backoff_ms);
xce_reconnect_fini(&rctx);
}
static void test_reconnect_fini_clears_endpoint(void)
{
reset_seam();
const char* h[] = { "ctrl1" };
const int p[] = { 9990 };
xce_cluster_config_t cfg;
make_cfg(&cfg, h, p, 1);
xce_reconnect_ctx_t rctx;
xce_reconnect_init(&rctx, &cfg);
xce_reconnect_step(&rctx); /* connect */
TEST_ASSERT_NOT_NULL(rctx.endpoint);
xce_reconnect_fini(&rctx);
TEST_ASSERT_NULL(rctx.endpoint);
}
/* ============================================================
* test group entry point
* ============================================================ */
void test_cluster(void)
{
/* Validation */
RUN_TEST(test_validate_rejects_null);
RUN_TEST(test_validate_rejects_zero_candidates);
RUN_TEST(test_validate_rejects_too_many_candidates);
RUN_TEST(test_validate_rejects_empty_ca_cert);
RUN_TEST(test_validate_rejects_empty_identity);
RUN_TEST(test_validate_rejects_bad_port);
RUN_TEST(test_validate_accepts_valid_single);
RUN_TEST(test_validate_accepts_max_candidates);
/* Candidate selection */
RUN_TEST(test_connect_cluster_returns_non_null_on_success);
RUN_TEST(test_connect_cluster_tries_second_on_first_fail);
RUN_TEST(test_connect_cluster_returns_null_if_all_fail);
RUN_TEST(test_connect_cluster_rejects_invalid_config);
/* Reconnect state machine */
RUN_TEST(test_reconnect_init_sets_idle_state);
RUN_TEST(test_reconnect_init_rejects_null_config);
RUN_TEST(test_reconnect_step_transitions_to_connected);
RUN_TEST(test_reconnect_step_enters_backoff_on_fail);
RUN_TEST(test_reconnect_backoff_advances);
RUN_TEST(test_reconnect_resets_backoff_on_connect);
RUN_TEST(test_reconnect_fini_clears_endpoint);
}
```
### meson.build additions for P2.5-8
```meson
# In libxce_test_src, add:
'testsuite/test_cluster.c',
```
### test.c additions for P2.5-8
```c
/* In src/lib/testsuite/test.c, add: */
extern void test_cluster(void);
/* In the tests[] array, add: */
{ "cluster", test_cluster },
```
### Commit
```bash
git add src/lib/testsuite/test_cluster.c \
src/lib/cluster_testseam.h \
src/lib/meson.build \
src/lib/testsuite/test.c
git commit -m "test: candidate selection, redirect, backoff, reconnect state machine (P2.5-8)"
```
---
## P2.5-7: CGo bridge in clib.go (DEFERRED)
**Status**: Deferred to a follow-up story, to be delivered alongside P2 replication.
**Rationale from design doc**: "Deploy P2.5-7 only alongside P2." Without P2 replication, a promoted standby has incomplete state — reconnecting the API server to it via the discovery path is misleading rather than helpful.
**When implementing**: Add `#include "xce/cluster.h"` to `backend/clib.go` preamble, expose `XceConnectCluster(conf *C.xce_cluster_config_t) *C.endpoint_t` wrapper, and call it from `RegisterToBackend()` instead of the current hardcoded dial. Parse `--peers` and `--ca-cert` from the API server's own startup config (already parsed into `BackendConfig`) and populate `C.xce_cluster_config_t` before the call.
---
## Build verification commands
```bash
# C library + tests
cd /worktrees/persistent_ha/src
nix develop --command bash -c "meson setup build --wipe && ninja -C build && ./build/lib/xcelib_test cluster"
# Go API server
cd /worktrees/persistent_ha/src/frontend
nix develop --command bash -c "make test"
# Specific Go test
cd /worktrees/persistent_ha/src/frontend/apiserver
nix develop --command bash -c "go test ./models/... -run TestJobStateStr -v"
```
---
## Implementation sequence
```
P2.5-1 → creates cluster.h (dependency for all C tasks)
P2.5-2 → creates cluster.c skeleton + test seam
P2.5-2a → wires cluster.c into meson build (build fails until here)
P2.5-8 → tests all pass after P2.5-1/2/2a
P2.5-3 → (confirmed by P2.5-8 — no extra commits if tests pass)
P2.5-5 → worker main.c (depends on cluster.h + parse_peers.c)
P2.5-6 → scheduler main.c (same pattern as worker)
P2.5-7a → Go models change (fully independent — run any time)
```
P2.5-7a can be committed any time (no C dependency). All other tasks depend on P2.5-1 being done first.
View git blame Copy permalink