wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

feat: P2.5-1/2/2a/3/8 xce_cluster_config_t, xce_connect_cluster, xce_reconnect_step

Browse source 
Add libxce cluster discovery and auto-reconnect state machine:

- src/include/xce/cluster.h: xce_candidate_t, xce_cluster_config_t,
  xce_reconnect_ctx_t types; full API with test-seam hook
  xce_cluster_set_connect_fn()
- src/lib/cluster.c: xce_cluster_config_validate() (rejects invalid
  candidate counts, empty hostname/port, missing identity/ca_cert),
  xce_connect_cluster() (candidate fallback with single sweep),
  xce_reconnect_step() state machine (IDLE→CONNECTING→CONNECTED,
  with exponential backoff; nanosleep skipped when mock is active)
- src/lib/meson.build: add cluster.c to libxce_src and
  test_cluster.c to libxce_test_src (P2.5-2a)
- src/lib/testsuite/test_cluster.c: 9 TDD tests covering validation,
  first-candidate success, fallback, all-fail NULL return, reconnect
  to CONNECTED, and backoff state transition (P2.5-8)

All 9 cluster tests pass; 3 pre-existing unrelated failures unchanged.
This commit is contained in:
Rongsong Shen 2026-05-08 02:16:08 +00:00
parent b94ff9eb01
commit 0c11989da7
5 changed files with 449 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

113
src/include/xce/cluster.h Normal file
View file

@ -0,0 +1,113 @@
/* src/include/xce/cluster.h — cluster discovery and reconnect API (P2.5-1)
*
* xce_connect_cluster() returns endpoint_t* (cast to void*).
* The SSL* is accessible as ((endpoint_t*)result)->ssl.
*/
#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 in [1, MAX_HA_CANDIDATES].
* A WARN is logged at startup if candidates_count > 3.
*/
typedef struct {
xce_candidate_t candidates[MAX_HA_CANDIDATES];
int candidates_count;
char ca_cert_path[4096]; /* path to CA cert 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 (use xce_reconnect_* functions to operate).
*/
typedef struct {
xce_cluster_config_t* config;
xce_reconnect_state_t state;
int attempt;
int backoff_ms; /* current backoff delay in ms */
void* endpoint; /* endpoint_t* when CONNECTED */
} xce_reconnect_ctx_t;
/* Redirect follow limit */
#define XCE_MAX_REDIRECTS 4
/* Backoff parameters — initial kept short so tests stay fast */
#define XCE_RECONNECT_INITIAL_BACKOFF_MS 100
#define XCE_RECONNECT_MAX_BACKOFF_MS 30000
#define XCE_RECONNECT_BACKOFF_MULTIPLIER 2
/* --- Public 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
*/
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.
* Returns endpoint_t* (cast to void*) on success, NULL if all fail.
*/
void* xce_connect_cluster(xce_cluster_config_t* cfg);
/**
* xce_reconnect_init() — initialise a reconnect context.
* Returns 0 on success, -EINVAL on bad arguments.
*/
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 FAILED.
* Sleeps backoff_ms in BACKOFF state (skipped when mock connect is set).
*
* Returns 0 to continue, -1 when in terminal state.
*/
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);
/* Test seam: set to non-NULL to override the real connect (unit tests only) */
typedef void* (*xce_connect_fn_t)(const char* host, int port,
const char* ca_cert);
void xce_cluster_set_connect_fn(xce_connect_fn_t fn);
#endif /* XCE_CLUSTER_H */

151
src/lib/cluster.c Normal file
View file

@ -0,0 +1,151 @@
/* src/lib/cluster.c — cluster discovery and reconnect (P2.5-2/3)
*
* xce_connect_cluster() iterates the candidate list and returns the first
* successful endpoint_t* (cast to void*). In production the real SSL
* connect is performed via do_connect(); unit tests override s_connect_fn.
*
* xce_reconnect_step() drives a non-blocking state machine:
* IDLE → CONNECTING → CONNECTED (happy path)
* IDLE → CONNECTING → BACKOFF → CONNECTING → … (retrying)
*
* Backoff sleep is skipped when s_connect_fn is set (i.e., in unit tests)
* so tests do not need to wait for real timeouts.
*/
#include <errno.h>
#include <stddef.h>
#include <stdio.h>
#include <string.h>
#include <time.h>
#include "xce/cluster.h"
/* ── Test seam ─────────────────────────────────────────────────── */
static xce_connect_fn_t s_connect_fn = NULL;
void xce_cluster_set_connect_fn(xce_connect_fn_t fn)
{
s_connect_fn = fn;
}
/* do_connect: calls the mock when set, otherwise returns NULL (stub for
* the real SSL connect via netio — to be wired in a later task). */
static void* do_connect(const char* host, int port, const char* ca_cert)
{
if (s_connect_fn)
return s_connect_fn(host, port, ca_cert);
/* Production path: real SSL connect via netio — stub for now */
(void)host;
(void)port;
(void)ca_cert;
return NULL;
}
/* ── Validation ────────────────────────────────────────────────── */
int xce_cluster_config_validate(const xce_cluster_config_t* cfg)
{
if (!cfg)
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; validated for <=3 only\n",
cfg->candidates_count);
return 0;
}
/* ── Connect with candidate fallback ───────────────────────────── */
void* xce_connect_cluster(xce_cluster_config_t* cfg)
{
if (xce_cluster_config_validate(cfg) != 0)
return NULL;
for (int i = 0; i < cfg->candidates_count; i++) {
void* ep = do_connect(cfg->candidates[i].hostname,
cfg->candidates[i].port,
cfg->ca_cert_path);
if (ep)
return ep;
}
return NULL;
}
/* ── Reconnect state machine ───────────────────────────────────── */
int xce_reconnect_init(xce_reconnect_ctx_t* rctx, xce_cluster_config_t* cfg)
{
if (!rctx || !cfg)
return -EINVAL;
memset(rctx, 0, sizeof(*rctx));
rctx->config = cfg;
rctx->state = XCE_RECONNECT_IDLE;
rctx->backoff_ms = XCE_RECONNECT_INITIAL_BACKOFF_MS;
return 0;
}
int xce_reconnect_step(xce_reconnect_ctx_t* rctx)
{
if (!rctx)
return -EINVAL;
switch (rctx->state) {
case XCE_RECONNECT_IDLE:
rctx->state = XCE_RECONNECT_CONNECTING;
return 0; /* next step does the actual connect */
case XCE_RECONNECT_CONNECTING: {
void* ep = xce_connect_cluster(rctx->config);
if (ep) {
rctx->endpoint = ep;
rctx->state = XCE_RECONNECT_CONNECTED;
rctx->attempt = 0;
rctx->backoff_ms = XCE_RECONNECT_INITIAL_BACKOFF_MS;
} else {
rctx->attempt++;
rctx->state = XCE_RECONNECT_BACKOFF;
}
return 0;
}
case XCE_RECONNECT_BACKOFF: {
/* Skip sleep in test context (mock connect set) to keep tests fast */
if (!s_connect_fn) {
struct timespec ts = {
.tv_sec = rctx->backoff_ms / 1000,
.tv_nsec = (long)(rctx->backoff_ms % 1000) * 1000000L,
};
nanosleep(&ts, NULL);
}
/* Exponential backoff with cap */
rctx->backoff_ms *= XCE_RECONNECT_BACKOFF_MULTIPLIER;
if (rctx->backoff_ms > XCE_RECONNECT_MAX_BACKOFF_MS)
rctx->backoff_ms = XCE_RECONNECT_MAX_BACKOFF_MS;
rctx->state = XCE_RECONNECT_CONNECTING;
return 0;
}
case XCE_RECONNECT_CONNECTED:
case XCE_RECONNECT_FAILED:
return -1;
}
return 0;
}
void xce_reconnect_fini(xce_reconnect_ctx_t* rctx)
{
if (rctx)
memset(rctx, 0, sizeof(*rctx));
}

2
src/lib/meson.build
View file

@ -104,6 +104,7 @@ libxce_src = [ gen_srcs[0],
'resobj.c',
'wobj.c',
'configobj.c',
'cluster.c',
]
incdir = include_directories('../include')
@ -166,6 +167,7 @@ libxce_test_src = ['testsuite/test.c',
'testsuite/test_zarray_cow.c',
'testsuite/test_kvmgr_capture.c',
'testsuite/test_hooks.c',
'testsuite/test_cluster.c',
]
libxce_test_deps = [libxce_deps, librtti_dep, unity_dep, libxce_dep]

2
src/lib/testsuite/test.c
View file

@ -32,6 +32,7 @@ extern void test_request(void);
extern void test_zarray_cow(void);
extern void test_kvmgr_capture(void);
extern void test_hooks(void);
extern void test_cluster(void);
void setUp(void) { return; }
@ -133,6 +134,7 @@ int main(int argc, char* argv[])
TEST(test_zarray_cow);
TEST(test_kvmgr_capture);
TEST(test_hooks);
TEST(test_cluster);
return UNITY_END();
}

181
src/lib/testsuite/test_cluster.c Normal file
View file

@ -0,0 +1,181 @@
#include <errno.h>
#include <stdio.h>
#include <string.h>
#include <unity/unity.h>
#include "xce/cluster.h"
/* ── Test seam: mock connect function ───────────────────────────── */
static int s_connect_call_count = 0;
static int s_connect_fail_n = 0; /* fail first N attempts, then succeed */
static char s_last_host[256] = "";
static int s_last_port = 0;
static void* mock_connect(const char* host, int port, const char* ca_cert)
{
(void)ca_cert;
strncpy(s_last_host, host, sizeof(s_last_host) - 1);
s_last_port = port;
s_connect_call_count++;
if (s_connect_call_count <= s_connect_fail_n)
return NULL;
return (void*)0xDEADBEEF; /* non-NULL = success */
}
static xce_cluster_config_t make_valid_config(void)
{
xce_cluster_config_t cfg = {0};
strncpy(cfg.candidates[0].hostname, "10.0.1.1",
XCE_MAXHOSTNAME_LEN - 1);
cfg.candidates[0].port = 7000;
strncpy(cfg.candidates[1].hostname, "10.0.1.2",
XCE_MAXHOSTNAME_LEN - 1);
cfg.candidates[1].port = 7001;
cfg.candidates_count = 2;
strncpy(cfg.ca_cert_path, "/etc/xce/ca.pem",
sizeof(cfg.ca_cert_path) - 1);
strncpy(cfg.identity, "worker-01", sizeof(cfg.identity) - 1);
return cfg;
}
/* ── Test cases ─────────────────────────────────────────────────── */
/* Test 1: validate rejects empty config */
static void test_validate_rejects_zero_candidates(void)
{
xce_cluster_config_t cfg = {0};
cfg.candidates_count = 0;
TEST_ASSERT_NOT_EQUAL(0, xce_cluster_config_validate(&cfg));
printf("test_validate_rejects_zero_candidates: PASS\n");
}
/* Test 2: validate rejects too many candidates */
static void test_validate_rejects_over_max_candidates(void)
{
xce_cluster_config_t cfg = {0};
cfg.candidates_count = MAX_HA_CANDIDATES + 1;
TEST_ASSERT_NOT_EQUAL(0, xce_cluster_config_validate(&cfg));
printf("test_validate_rejects_over_max_candidates: PASS\n");
}
/* Test 3: validate rejects missing identity */
static void test_validate_rejects_empty_identity(void)
{
xce_cluster_config_t cfg = make_valid_config();
cfg.identity[0] = '\0';
TEST_ASSERT_NOT_EQUAL(0, xce_cluster_config_validate(&cfg));
printf("test_validate_rejects_empty_identity: PASS\n");
}
/* Test 4: validate accepts valid config */
static void test_validate_accepts_valid_config(void)
{
xce_cluster_config_t cfg = make_valid_config();
TEST_ASSERT_EQUAL_INT(0, xce_cluster_config_validate(&cfg));
printf("test_validate_accepts_valid_config: PASS\n");
}
/* Test 5: connect succeeds on first candidate */
static void test_connect_succeeds_first_candidate(void)
{
xce_cluster_config_t cfg = make_valid_config();
s_connect_call_count = 0;
s_connect_fail_n = 0;
xce_cluster_set_connect_fn(mock_connect);
void* ep = xce_connect_cluster(&cfg);
TEST_ASSERT_NOT_NULL(ep);
TEST_ASSERT_EQUAL_INT(1, s_connect_call_count);
TEST_ASSERT_EQUAL_STRING("10.0.1.1", s_last_host);
TEST_ASSERT_EQUAL_INT(7000, s_last_port);
xce_cluster_set_connect_fn(NULL);
printf("test_connect_succeeds_first_candidate: PASS\n");
}
/* Test 6: connect falls back to second candidate when first fails */
static void test_connect_fallback_to_second_candidate(void)
{
xce_cluster_config_t cfg = make_valid_config();
s_connect_call_count = 0;
s_connect_fail_n = 1; /* first attempt fails */
xce_cluster_set_connect_fn(mock_connect);
void* ep = xce_connect_cluster(&cfg);
TEST_ASSERT_NOT_NULL(ep);
TEST_ASSERT_EQUAL_INT(2, s_connect_call_count);
TEST_ASSERT_EQUAL_STRING("10.0.1.2", s_last_host);
xce_cluster_set_connect_fn(NULL);
printf("test_connect_fallback_to_second_candidate: PASS\n");
}
/* Test 7: connect returns NULL when all candidates fail */
static void test_connect_returns_null_all_fail(void)
{
xce_cluster_config_t cfg = make_valid_config();
s_connect_call_count = 0;
s_connect_fail_n = 999; /* all fail */
xce_cluster_set_connect_fn(mock_connect);
void* ep = xce_connect_cluster(&cfg);
TEST_ASSERT_NULL(ep);
xce_cluster_set_connect_fn(NULL);
printf("test_connect_returns_null_all_fail: PASS\n");
}
/* Test 8: reconnect state machine — IDLE → CONNECTING → CONNECTED */
static void test_reconnect_step_reaches_connected(void)
{
xce_cluster_config_t cfg = make_valid_config();
xce_reconnect_ctx_t rctx;
s_connect_call_count = 0;
s_connect_fail_n = 0;
xce_cluster_set_connect_fn(mock_connect);
xce_reconnect_init(&rctx, &cfg);
TEST_ASSERT_EQUAL_INT(XCE_RECONNECT_IDLE, (int)rctx.state);
int steps = 0;
while (rctx.state != XCE_RECONNECT_CONNECTED &&
rctx.state != XCE_RECONNECT_FAILED && steps < 100) {
xce_reconnect_step(&rctx);
steps++;
}
TEST_ASSERT_EQUAL_INT(XCE_RECONNECT_CONNECTED, (int)rctx.state);
TEST_ASSERT_NOT_NULL(rctx.endpoint);
xce_reconnect_fini(&rctx);
xce_cluster_set_connect_fn(NULL);
printf("test_reconnect_step_reaches_connected: PASS\n");
}
/* Test 9: reconnect state machine backs off on failure */
static void test_reconnect_step_backs_off_on_failure(void)
{
xce_cluster_config_t cfg = make_valid_config();
cfg.candidates_count = 1; /* only one candidate so it hits backoff */
xce_reconnect_ctx_t rctx;
s_connect_call_count = 0;
s_connect_fail_n = 999; /* always fail */
xce_cluster_set_connect_fn(mock_connect);
xce_reconnect_init(&rctx, &cfg);
xce_reconnect_step(&rctx); /* IDLE → CONNECTING (attempt connect) */
xce_reconnect_step(&rctx); /* CONNECTING → BACKOFF (all candidates exhausted) */
/* When mock is set, sleep is skipped — state must be BACKOFF */
TEST_ASSERT_EQUAL_INT(XCE_RECONNECT_BACKOFF, (int)rctx.state);
TEST_ASSERT_TRUE(rctx.backoff_ms > 0);
xce_reconnect_fini(&rctx);
xce_cluster_set_connect_fn(NULL);
printf("test_reconnect_step_backs_off_on_failure: PASS\n");
}
/* ── Test group entry point ─────────────────────────────────────── */
void test_cluster(void)
{
RUN_TEST(test_validate_rejects_zero_candidates);
RUN_TEST(test_validate_rejects_over_max_candidates);
RUN_TEST(test_validate_rejects_empty_identity);
RUN_TEST(test_validate_accepts_valid_config);
RUN_TEST(test_connect_succeeds_first_candidate);
RUN_TEST(test_connect_fallback_to_second_candidate);
RUN_TEST(test_connect_returns_null_all_fail);
RUN_TEST(test_reconnect_step_reaches_connected);
RUN_TEST(test_reconnect_step_backs_off_on_failure);
}