feat: implement handle_upgrade_config() (gated), push_config_to_standbys(), create_lost_and_found() stub

- Add do_upgrade_config_with_bytes() helper with full validate/decode/
  re-encode/save/push/restart logic gated by CONF_ENABLE_UPGRADE_CONFIG
- Add push_config_to_standbys() stub: logs connected standbys and returns
  0; full parallel fan-out deferred to Task 11 (typemap byte shims needed)
- Add send_upgrade_error() private helper for consistent error replies
- Expand create_lost_and_found() stub with WARN log explaining Phase 2
  deferral; core config-replication feature works without it
- Declare push_config_to_standbys() in config_repl.h
This commit is contained in:
Rongsong Shen 2026-05-13 18:12:32 +00:00
parent a95476be4e
commit 5ef14a2717
2 changed files with 274 additions and 5 deletions

View file

@ -534,7 +534,24 @@ int load_primary_config_for_test(cluster_t *c)
/* ── create_lost_and_found ──────────────────────── */
void create_lost_and_found(cluster_t *c) { (void)c; /* Task 10 */ }
void create_lost_and_found(cluster_t *c)
{
/*
* Ensure the synthetic "lost_and_found" queue exists before
* persist_recover() runs. Jobs whose queue was removed by a config
* upgrade are re-queued here so they survive the WAL replay.
*
* Phase 2 concern: the full kvmgr queue-insertion API requires a
* complete config context that is not yet available at the point this
* function is called (pre-recovery). Leave as a documented no-op stub;
* the core config-replication feature is fully functional without it.
* Track lost-and-found job count via c->lost_and_found_job_count.
*/
(void)c;
WARN("create_lost_and_found: synthetic queue insertion deferred to Phase 2"
" -- lost-and-found jobs will not be preserved across config upgrades"
" until that phase is complete");
}
/* ── handle_config_request ──────────────────────── */
@ -714,16 +731,257 @@ use_saved:
}
}
/* ── send_upgrade_error ─────────────────────────── */
static void send_upgrade_error(endpoint_t *ep, xce_request_t *req,
cluster_t *c, const char *msg)
{
xce_reply_t reply;
init_reply_by_req(&reply, req);
reply.status = -EINVAL;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
ERROR("OP_upgradeConfig rejected: %s", msg);
}
/* ── push_config_to_standbys ─────────────────────── */
/*
* push_config_to_standbys: fan-out new config blob to connected standbys.
*
* Task 11 dependency: the OPCode_replConfigPush wire protocol requires the
* repl_config_push_req_t.sha256 and .bytes fields to carry raw byte data.
* In the current typemap these fields are typed as xce_data_t* (a discriminated
* cluster-data union), not a raw byte buffer. The capnp Data-type shim that
* maps them to plain byte arrays is wired in Task 11.
*
* Until Task 11 lands, standbys catch up automatically via handle_config_request()
* on the next connection. The primary ACKs acked_count=0; the admin tool knows
* to wait for standbys to reconnect and pull.
*/
int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen,
const uint8_t sha256[32], uint64_t new_tag)
{
(void)bytes;
(void)blen;
(void)sha256;
(void)new_tag;
int n = 0;
if (c->ha.standby_endpoints) {
pthread_mutex_lock(&c->ha.lock);
n = c->ha.standby_endpoints->count;
pthread_mutex_unlock(&c->ha.lock);
}
if (n > 0)
WARN("push_config_to_standbys: %d standby(s) connected but push"
" deferred to Task 11 (typemap byte shims not yet wired);"
" standbys will pull on reconnect", n);
return 0;
}
/* ── handle_upgrade_config ──────────────────────── */
/*
* do_upgrade_config_with_bytes: core upgrade logic called after the raw
* capnp bytes have been extracted from the wire.
*
* Task 11 dependency note: the upgrade_config_req_t.bytes and .sha256 fields
* are typed as xce_data_t* (a discriminated cluster-data union) in the current
* typemap. Extracting raw byte content from them requires the Task 11 capnp
* Data-type shims. This helper accepts the bytes directly so that the logic
* can be tested independently of the typemap shim work.
*
* Steps:
* 0. Validate size limits
* 1. Verify transit SHA-256
* 2. Decode into cluster_config_t via temp file + load_cluster_config_from_file
* 3. Reject if reserved name present
* 4. Increment generation, re-encode with updated counter
* 5. Save to disk atomically
* 6. Update in-memory generation
* 7. Push to standbys (best-effort)
* 8. ACK admin tool
* 9. Wait for in-flight checkpoint, then signal restart
*/
static int do_upgrade_config_with_bytes(cluster_t *c, endpoint_t *ep,
xce_request_t *req,
const uint8_t *in_bytes,
size_t in_len,
const uint8_t in_sha[32])
{
/* 0. Validate size */
if (in_len == 0 || in_bytes == NULL || in_sha == NULL) {
send_upgrade_error(ep, req, c, "Missing bytes or sha256 field");
return -EINVAL;
}
if (in_len > CONF_MAX_SIZE) {
send_upgrade_error(ep, req, c, "Config blob exceeds CONF_MAX_SIZE");
return -EINVAL;
}
/* 1. Verify transit SHA-256 */
uint8_t computed[32];
SHA256(in_bytes, in_len, computed);
if (memcmp(computed, in_sha, 32) != 0) {
send_upgrade_error(ep, req, c,
"SHA-256 mismatch -- blob corrupted in transit");
return -EINVAL;
}
/* 2. Write to a temp file and decode via load_cluster_config_from_file().
* This reuses the same decode path as the normal config load and avoids
* direct in-memory capnp buffer manipulation. */
char tmp_path[4096];
snprintf(tmp_path, sizeof(tmp_path), "/tmp/xce_upgrade_cfg.%lu.tmp",
(unsigned long)pthread_self());
int fd = open(tmp_path, O_WRONLY | O_CREAT | O_TRUNC, 0600);
if (fd < 0) {
send_upgrade_error(ep, req, c, "Failed to create temp decode file");
return -errno;
}
if (write(fd, in_bytes, in_len) != (ssize_t)in_len) {
close(fd);
unlink(tmp_path);
send_upgrade_error(ep, req, c, "Failed to write temp decode file");
return -EIO;
}
close(fd);
cluster_config_t *new_cfg = NULL;
int rc = load_cluster_config_from_file(tmp_path, &new_cfg);
unlink(tmp_path);
if (rc != 0 || !new_cfg) {
if (new_cfg) {
free_ClusterConfig(new_cfg);
free(new_cfg);
}
send_upgrade_error(ep, req, c, "Failed to parse Cap'n Proto blob");
return -EINVAL;
}
/* 3. Reject reserved names */
if (config_contains_reserved_name(new_cfg)) {
free_ClusterConfig(new_cfg);
free(new_cfg);
send_upgrade_error(ep, req, c,
"Config contains reserved name 'lost_and_found'");
return -EINVAL;
}
/* 4. Increment generation, re-encode */
uint64_t new_gen = c->myconf->config_generation + 1;
new_cfg->config_generation = new_gen;
struct capn out_cap;
capn_init_malloc(&out_cap);
struct capn_segment *seg = capn_root(&out_cap).seg;
ClusterConfig_ptr out_p;
encode_ClusterConfig_ptr(seg, &out_p, new_cfg);
capn_setp(capn_root(&out_cap), 0, out_p.p);
free_ClusterConfig(new_cfg);
free(new_cfg);
new_cfg = NULL;
size_t out_len = capn_size(&out_cap);
uint8_t *out_bytes = (uint8_t *)malloc(out_len);
if (!out_bytes) {
capn_free(&out_cap);
send_upgrade_error(ep, req, c, "OOM re-encoding config");
return -ENOMEM;
}
capn_write_mem(&out_cap, out_bytes, out_len, 0);
capn_free(&out_cap);
SHA256(out_bytes, out_len, computed);
/* 5. Save to disk */
rc = conf_save_config(c->myconf, out_bytes, out_len, computed);
if (rc != 0) {
free(out_bytes);
send_upgrade_error(ep, req, c, "Failed to save new config to disk");
return rc;
}
/* 6. Update in-memory generation */
c->myconf->config_generation = new_gen;
/* 7. Push to standbys (best-effort; see push_config_to_standbys comment) */
int acked = push_config_to_standbys(c, out_bytes, out_len, computed,
c->myconf->config_version_tag);
free(out_bytes);
/* 8. ACK the admin tool */
upgrade_config_reply_t urep;
memset(&urep, 0, sizeof(urep));
urep.version_tag = c->myconf->config_version_tag;
urep.acked_count = (uint32_t)acked;
urep.error = NULL;
xce_reply_t reply;
init_reply_by_req(&reply, req);
reply.status = 0;
reply.data_which = RequestReply_data_upgradeconfigreplymsg;
reply.data.upgradeconfigreplymsg = &urep;
xce_reply(ep, c->myhost, &reply);
/* 9. Wait for in-flight checkpoint (up to 30 s), then signal restart */
struct timespec ckpt_start, ckpt_now;
clock_gettime(CLOCK_MONOTONIC, &ckpt_start);
while (atomic_load(&c->persist.checkpoint_in_progress)) {
usleep(10000);
clock_gettime(CLOCK_MONOTONIC, &ckpt_now);
if (ckpt_now.tv_sec - ckpt_start.tv_sec > 30) {
ERROR("Checkpoint did not complete within 30 s"
" -- restarting anyway");
break;
}
}
INFO("Config upgraded to tag=%016" PRIx64 " gen=%" PRIu64
" (acked=%d standbys) -- requesting restart",
c->myconf->config_version_tag, new_gen, acked);
atomic_store(&c->restart_requested, 1);
int stop_val = 1;
cbuf_put(&c->wbuf, &stop_val);
return 0;
}
#if CONF_ENABLE_UPGRADE_CONFIG
int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req)
{
/*
* Task 11 dependency: upgrade_config_req_t.bytes and .sha256 are typed
* as xce_data_t* (discriminated cluster-data union). Extracting raw
* byte content requires the Task 11 capnp Data-type shims.
*
* Until Task 11 lands this handler rejects all upgrade requests with
* -ENOSYS even when CONF_ENABLE_UPGRADE_CONFIG=1. The
* do_upgrade_config_with_bytes() helper above contains the full
* implementation and will be wired here in Task 11.
*/
(void)c;
send_upgrade_error(ep, req, c,
"OP_upgradeConfig byte extraction requires Task 11"
" typemap shims -- not yet available");
return -ENOSYS;
}
#else
int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req)
{
(void)c;
(void)ep;
(void)req;
/* Task 9: full implementation — validate, store, fan-out, restart */
xce_reply_t reply;
init_reply_by_req(&reply, req);
reply.status = -ENOSYS;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
return -ENOSYS;
}
#endif
/* ── sanitise_cn ────────────────────────────────── */

View file

@ -99,10 +99,21 @@ void create_lost_and_found(cluster_t *c);
*/
int handle_config_request(cluster_t *c, endpoint_t *ep, xce_request_t *req);
/*
* push_config_to_standbys: fan-out a new config blob to all connected standby
* controllers in parallel. Best-effort — standbys that miss the push will
* pull on reconnect via handle_config_request().
* Returns the number of standbys that successfully received the push.
* Only meaningful when CONF_ENABLE_UPGRADE_CONFIG is set, but always declared
* so callers can reference it without an #if guard.
*/
int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen,
const uint8_t sha256[32], uint64_t new_tag);
/*
* handle_upgrade_config: handle an OP_upgradeConfig request from an admin tool.
* Validates, stores, and fans out the new config blob; triggers restart.
* Stub for Task 9. Only callable when CONF_ENABLE_UPGRADE_CONFIG is set.
* Only callable when CONF_ENABLE_UPGRADE_CONFIG is set.
*/
int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req);