From 7f8c07ba85e4d1022403669acda2f5bf5da1e1a7 Mon Sep 17 00:00:00 2001 From: "lei.chen" Date: Fri, 14 Aug 2026 07:29:32 +0000 Subject: [PATCH] =?UTF-8?q?fix(scheduler):=20bug-1513=20=E2=80=94=20rejoin?= =?UTF-8?q?=20crash=20root-caused=20to=203=20rejoin=20re-push=20bugs,=20no?= =?UTF-8?q?t=20a=20lib=20race?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The "libxce reconnect heap corruption" (ASAN bad-free, free(): invalid pointer) was NOT a libxce bug — xce_ctx_reconnect is fine. The crashes are in the scheduler's rejoin state re-sync path (config/dataset re-push), fixed with the reconnect/rejoin feature now integrated: 1. group.c makeGroup refresh: FREEUP(m->u.name) raw-freed shared interned xstrings (makeString is a global refcounted table); the correct release is dereference(), matching group_killer. → dereference(). 2. limit.c limit_update_config: after key_add returns -EEXIST (rejoin re-push — limit name already present) it called kill_limit(limit), but xkvmgr_add's vclear already freed the limit via the kvmgr datakiller (=kill_limit). → skip on -EEXIST (idempotent re-push). 3. handler.c/scheduler.c: the config push was applied inline on the vcpu thread (sched_init_config adds queues/plugin queues) while job updates flow through pend_updates on the main loop — concurrent rejoin re-push let fcfs_update_job see a NULL j->queue (SIGSEGV). Route the config push through pend_updates (XCE_SCHED_OP_UPDATE_CONFIG, killer case already existed) and process config requests FIRST in sched_update via a staging queue so jobs always see their queues/resources. Verified: 295 lib unit tests pass under ASAN (incl. 2 new group-refresh regression tests), scheduler survives 3 consecutive controller restarts (each rejoin ok), full job lifecycle (submit→dispatch→DONE) on the cluster. Also fixed stale build deps (libxce_deps missing librtti_dep, scheduler_deps missing ssl_dep/c_capnproto_dep, unity include path). Co-Authored-By: Claude --- docs/left_scheduler_work.md | 16 +++--- src/lib/group.c | 6 +- src/lib/meson.build | 9 ++- src/lib/testsuite/test_group.c | 95 ++++++++++++++++++++++++++++++- src/scheduler/include/scheduler.h | 3 + src/scheduler/meson.build | 4 +- src/scheduler/src/handler.c | 34 ++++++++--- src/scheduler/src/main.c | 15 +++++ src/scheduler/src/plugins/limit.c | 12 +++- src/scheduler/src/scheduler.c | 86 +++++++++++++++++++++++++++- 10 files changed, 253 insertions(+), 27 deletions(-) diff --git a/docs/left_scheduler_work.md b/docs/left_scheduler_work.md index a35833c..8c791d0 100644 --- a/docs/left_scheduler_work.md +++ b/docs/left_scheduler_work.md @@ -94,13 +94,15 @@ spec 声称 36/36 任务完成，但 re-audit（research.md）发现缺陷。来 - **apiserver rejoin**（clib.go `StartHeartbeat` + `xce_backend_rejoin`）：心跳从失败转成功时显式 rejoin（HostRole_apiserver），恢复 push 通知。实测：controller 重启后日志 `backend re-joined controller after reconnect`。 - **apiserver fail-fast 电路熔断（Req 4，commit 2bbe894）**：health.go 原子熔断 + `callBackend` 包 4 个写调用。degraded 时立即返回 `backend controller unavailable`，不调 lib；传输错误或 XCE NOMASTER/REDIRECT/NOT_READY/STANDBY 触发熔断；语义错误不触发。实测：杀 controller 后第 2 次提交 ~11ms 返回 fail-fast。7 单测。 - **⚠️ 已知缺口**：**apiserver 心跳恢复在 controller 完整死亡+重启后不总是可靠**——lib transport reconnect 未稳定重建到新 controller（与 scheduler reconnect 同类，task #13）。当前需手动重启 apiserver 恢复（clean rejoin 正常）。 -- **❌ scheduler reconnect 被真实 heap corruption 阻塞（未合入，已回退）**： - - scheduler 的 rejoin（`xcecli_set_reconnect_handler` flag + 主循环 `xcecli_join`）**功能上能重连并 rejoin**（实测收到 config push），但 rejoin 后 ~3-8s **崩溃**。 - - **诊断（2026-08-14 深入）**：ASAN 报 `bad-free`（释放 60 字节区域偏移 56 处的指针，`...66b8`，分配线程 T4/释放线程 T5，**全栈在 loader**）；**debug 非 ASAN 构建同样崩**（`free(): invalid pointer` + core，gdb 显示 malloc_printerr 后栈被破坏）——**真实堆损坏，非 ASAN 假象**。 - - 尝试的修复（均无效）：`xce_ctx_reconnect` 先 `iomgr_unmonitor` 再 `endpoint_fini`（对齐 xce_ctx_stop 的安全拆除，正确但不充分）。 - - **候选根因**：(a) rejoin 触发 config 重推 → `sched_init_config` 非幂等重跑（sharedInst 的 MOVE_PTR/zarray_put killer 覆盖、queues/resources 重加）；(b) lib reconnect 与 vtask 并发处理旧 endpoint；(c) 重新同步时 scheduler 本地状态与重推的 merge 缺陷。栈被破坏无法干净回溯，需 valgrind/隔离测试定位。 - - **修法**：独立深层调试任务（scheduler 状态重同步幂等性 + lib reconnect/vtask 生命周期）。为保持 scheduler 稳定已回退全部 reconnect 改动。 -- 剩余任务（scheduler reconnect + 多 controller E2E + apiserver 电路熔断等）见 spec。 +- **✅ scheduler reconnect 已修复（2026-08-14 根因定位 + 集群验证）**： + - **关键结论**：所谓 "libxce reconnect heap corruption" **不是 lib 竞态**——是 scheduler rejoin 状态重同步的 3 个真实 bug（均在 config/dataset 重推路径）。 + - **根因 1（group.c:234/247）**：`makeGroup` 刷新已有 group 的成员时 `FREEUP(m->u.name)` **raw-free 了共享 interned xstring**（`make_g_member` 用 `makeString` 建的是全局引用计数共享串，多个 group 共享同一指针）。正确释放是 `dereference()`（`group_killer` 就这么做）。ASAN 报 `bad-free`/`free(): invalid pointer` 即源于此。 + - **根因 2（limit.c:76）**：`limit_update_config` 在 `key_add` 返回 -EEXIST（rejoin 重推，limit 名已存在）后调用 `kill_limit(limit)`，但 `xkvmgr_add` 的 vclear 已用 kvmgr datakiller（=kill_limit）释放该 limit → **double-free/UAF**（ASAN：56 字节 limit 的 READ of size 8）。修复：EEXIST 时 skip（re-push 幂等）。 + - **根因 3（handler.c + scheduler.c）**：config push 在 vcpu 线程**内联**执行 `sched_init_config`（加 queues/plugin 队列），而 job 更新走 pend_updates 由主循环处理——**rejoin 重推时两者并发**，`fcfs_update_job` 的 `j->queue->name` 可能在 config 建好队列前取到 NULL → 主线程 SIGSEGV。修复：config push **改走 pend_updates**（`XCE_SCHED_OP_UPDATE_CONFIG`），`sched_update` 里 **config 请求先处理**（staging 队列）再处理 job/worker 更新，串行化顺序。 + - **验证**：295 个 lib 单测全过（ASAN，含 2 个新 group 刷新回归测试）；集群上 scheduler 连续 3 次 controller 重启全部存活并 `rejoin ok`；完整作业生命周期 `sleep 30` 提交→派发到 t1→**DONE**。 + - **附带修复**：`libxce_deps` 缺 `librtti_dep`（group.c 编译不过）、`scheduler_deps` 缺 `ssl_dep`/`c_capnproto_dep`、unity include 只给 `` 路径——均已在 meson.build 修复。 + - **apiserver 恢复缺口仍在（task #12）**：controller 完整死亡+重启后 apiserver 心跳恢复不可靠，电路熔断保持 degraded，需手动重启 apiserver（本次验证即遇到：job 提交报 `backend controller unavailable (reconnecting)`，重启 apiserver 后恢复）。 +- 剩余任务（多 controller E2E + apiserver 自动恢复等）见 spec。 ### P2.5 - **来源**：`product.md:35`、`docs/superpowers/plans/2026-05-07-p25-discovery.md` diff --git a/src/lib/group.c b/src/lib/group.c index b4c70e2..6bf5752 100644 --- a/src/lib/group.c +++ b/src/lib/group.c @@ -231,7 +231,9 @@ xce_group_t* makeGroup(var mgr, xce_group_config_t* config, continue; } if (m->is_group == 0) { - FREEUP(m->u.name); + /* member names are refcounted interned xstrings shared across + * groups — release the ref, never free() (bug-1513) */ + dereference(m->u.name); } FREEUP(m); @@ -244,7 +246,7 @@ xce_group_t* makeGroup(var mgr, xce_group_config_t* config, continue; } if (m->is_group == 0) { - FREEUP(m->u.name); + dereference(m->u.name); } FREEUP(m); diff --git a/src/lib/meson.build b/src/lib/meson.build index 6715cb5..a7182d4 100644 --- a/src/lib/meson.build +++ b/src/lib/meson.build @@ -122,7 +122,7 @@ tinylog_dep = dependency('tinylog') unity_dep = dependency('unity') ssl_dep = dependency('openssl') -libxce_deps = [c_capnproto_dep, ssl_dep, tinylog_dep, thread_dep] +libxce_deps = [c_capnproto_dep, ssl_dep, tinylog_dep, thread_dep, librtti_dep] #libxce_shared = shared_library('xce', # libxce_src, @@ -193,7 +193,12 @@ libxce_test_src = ['testsuite/test.c', libxce_test_deps = [libxce_deps, librtti_dep, unity_dep, libxce_dep] libxce_test = executable('xcelib_test', libxce_test_src, - include_directories: [incdir, build_inc], + # unity.pc's Cflags only adds -I${includedir}/unity, so + # resolves but (used by several + # suites) does not — add the parent includedir as well. + include_directories: [incdir, build_inc, + include_directories( + unity_dep.get_variable('includedir'))], dependencies: libxce_test_deps, c_args: ['-DXCE_TESTING'], install: true) diff --git a/src/lib/testsuite/test_group.c b/src/lib/testsuite/test_group.c index 0943835..58416a6 100644 --- a/src/lib/testsuite/test_group.c +++ b/src/lib/testsuite/test_group.c @@ -255,12 +255,103 @@ static void test_nest2(void) destroy(mgr); } +/* makeGroup refresh (same group name re-config'd) must release an existing + * member's interned name with dereference(), not free(). The member name is a + * refcounted xstring shared across groups — a raw free() leaves the other + * holder dangling and corrupts the xstring table (bug-1513: scheduler + * reconnect crash, free(): invalid pointer from makeGroup). */ +static void test_refresh_members(void) +{ + var mgr; + xce_group_config_t gdata = { + .name = "g1", + .members = { .count = 2, .data = (char*[]) { "rshen", "all" } } + }; + + xce_group_t* g = NULL; + int rc = 0; + + groupmgr_init(&mgr, 16, 16); + + /* A second group shares the SAME interned "rshen" string with g1. */ + xce_group_config_t gdata2 = { + .name = "g2", + .members = { .count = 1, .data = (char*[]) { "rshen" } } + }; + xce_group_t* g2 = makeGroup(mgr, &gdata2, NULL); + + TEST_ASSERT_TRUE(g2 != NULL); + + g = makeGroup(mgr, &gdata, NULL); + + TEST_ASSERT_TRUE(g != NULL); + + /* Refresh g1 with the same config — hits the existing-group path. */ + g = makeGroup(mgr, &gdata, NULL); + + TEST_ASSERT_TRUE(g != NULL); + + /* Both groups still resolve "rshen" — the shared string must survive. */ + rc = is_in_group(g, "rshen"); + + TEST_ASSERT_TRUE(rc > 0); + + rc = is_in_group(g2, "rshen"); + + TEST_ASSERT_TRUE(rc > 0); + + destroy(mgr); +} + +static void test_refresh_ex_members(void) +{ + var mgr; + xce_group_config_t gdata = { + .name = "g1", + .members = { .count = 1, .data = (char*[]) { "all" } }, + .ex_members = { .count = 1, .data = (char*[]) { "rshen" } } + }; + + xce_group_t* g = NULL; + int rc = 0; + + groupmgr_init(&mgr, 16, 16); + + /* g2 shares "rshen" with g1's ex_members. */ + xce_group_config_t gdata2 = { + .name = "g2", + .members = { .count = 1, .data = (char*[]) { "rshen" } } + }; + xce_group_t* g2 = makeGroup(mgr, &gdata2, NULL); + + TEST_ASSERT_TRUE(g2 != NULL); + + g = makeGroup(mgr, &gdata, NULL); + + TEST_ASSERT_TRUE(g != NULL); + + /* Refresh g1 — hits the ex_members cleanup path. */ + g = makeGroup(mgr, &gdata, NULL); + + TEST_ASSERT_TRUE(g != NULL); + + rc = is_in_group(g, "rshen"); + + TEST_ASSERT_TRUE(rc == 0); + + rc = is_in_group(g2, "rshen"); + + TEST_ASSERT_TRUE(rc > 0); + + destroy(mgr); +} + void test_group(void) { printf("\n>> group test\n\n"); xstring_table_init(139); - + RUN_TEST(test_all); RUN_TEST(test_empty); RUN_TEST(test_all_1); @@ -271,6 +362,8 @@ void test_group(void) RUN_TEST(test_exclude); RUN_TEST(test_nest); RUN_TEST(test_nest2); + RUN_TEST(test_refresh_members); + RUN_TEST(test_refresh_ex_members); xstring_table_fini(); } diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index 80e5ffc..064f70e 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -117,6 +117,8 @@ typedef struct { _Atomic int session_active; /* 1 during scheduling session, 0 between */ _Atomic int eval_in_progress; /* 1 while eval_select loop runs */ _Atomic int staging_merge_complete; /* 1 after staging merge completes */ + _Atomic int needs_rejoin; /* 1 when the lib reconnected and the + * scheduler session must be re-joined */ select_eval_ctx_t* sel_eval_ctx; /* session-level memoisation context; NULL between sessions */ xce_cluster_config_t cluster_cfg; /* HA candidate list — empty means single-host mode */ } xce_scheduler_t; @@ -129,6 +131,7 @@ extern int sched_update(xce_scheduler_t*); extern void sched_go(xce_scheduler_t*); extern void sched_begin_session(xce_scheduler_t*); extern void sched_end_session(xce_scheduler_t*); +extern int sched_rejoin_if_needed(xce_scheduler_t*); extern xce_sched_job_t* sched_get_job(xce_scheduler_t*, uint64_t); extern int sched_requeue_pend_job(xce_scheduler_t*, uint64_t); extern xce_sched_queue_t* sched_get_queue(xce_scheduler_t*, char*); diff --git a/src/scheduler/meson.build b/src/scheduler/meson.build index ccc3e8b..9bd6b99 100644 --- a/src/scheduler/meson.build +++ b/src/scheduler/meson.build @@ -22,7 +22,9 @@ scheduler_deps = [ librtti_dep, math_dep, tinylog_dep, - thread_dep + thread_dep, + ssl_dep, + c_capnproto_dep ] scheduler = executable('xce_scheduler', diff --git a/src/scheduler/src/handler.c b/src/scheduler/src/handler.c index 9436538..5e30ee4 100644 --- a/src/scheduler/src/handler.c +++ b/src/scheduler/src/handler.c @@ -592,17 +592,33 @@ int scheduler_req_handler(xce_request_t* request, void* args) switch (request->opcode) { case OPCode_push: switch (request->body_which) { - case Request_body_configpushmsg: - /* configuration packet, it will only arrive when scheduler has - * register itself to controller, we can handle it synchronized - */ - rc = sched_init_config( - sched, request->body.configpushmsg->data.conf4sched); - if (rc >= 0) { - /* configuration will be saved in sched->config */ - request->body.configpushmsg->data.conf4sched = NULL; + case Request_body_configpushmsg: { + /* Configuration packet. Route it through the pend-update queue so + * sched_init_config runs in the MAIN loop, serialised with the + * job/worker/queue updates it co-arrives with on a controller + * rejoin (bug-1513). Applying it inline on the vcpu thread races + * sched_update's job processing (fcfs_update_queue's ordq_put vs + * fcfs_update_job's ordq_peek on the same plugin queues), and the + * queues may not exist yet when the job update looks them up → + * NULL j->queue → SIGSEGV. */ + xce_sched_update_request_t* req; + xce_conf4sched_t* config = request->body.configpushmsg->data.conf4sched; + + req = (xce_sched_update_request_t*)calloc(1, sizeof(*req)); + if (req == NULL) { + rc = -ENOMEM; + break; } + + req->op = XCE_SCHED_OP_UPDATE_CONFIG; + req->u.config = config; + enq_pend_update(sched, req); + + /* Ownership of config moved to the queued request. */ + request->body.configpushmsg->data.conf4sched = NULL; + rc = 0; break; + } case Request_body_datamsg: rc = sched_handle_dataset(sched, request->body.datamsg); break; diff --git a/src/scheduler/src/main.c b/src/scheduler/src/main.c index dcb5132..2d7a58c 100644 --- a/src/scheduler/src/main.c +++ b/src/scheduler/src/main.c @@ -1,5 +1,6 @@ #include #include +#include #include #include #include @@ -156,7 +157,21 @@ int main(int argc, char* argv[]) INFO("xce scheduler is running"); + int hb_was_down = 0; + while (1) { + sched_rejoin_if_needed(&sched); + + { + int hb = xcecli_heartbeat(sched.apictx, 1); + if ((hb == 0) && hb_was_down) { + INFO("controller connection restored — requesting scheduler rejoin"); + atomic_store_explicit(&sched.needs_rejoin, 1, + memory_order_release); + } + hb_was_down = (hb < 0); + } + sched_update(&sched); sched_begin_session(&sched); sched_go(&sched); diff --git a/src/scheduler/src/plugins/limit.c b/src/scheduler/src/plugins/limit.c index 326b317..168ee02 100644 --- a/src/scheduler/src/plugins/limit.c +++ b/src/scheduler/src/plugins/limit.c @@ -73,8 +73,16 @@ static int limit_update_config(xce_sched_plugin_t* plugin, unlock(privdata->limits); if (rc < 0) { - kill_limit(limit); - + if (rc == -EEXIST) { + /* Rejoin re-push: the limit name already exists from the + * first config, and xkvmgr_add already rolled back — vclear + * ran the kvmgr's datakiller (kill_limit) and freed this + * freshly-built duplicate. Never free it again here (that + * double-free was the rejoin heap-use-after-free crash, + * bug-1513). The existing limit is still valid — skip and + * keep the config push idempotent. */ + continue; + } return rc; } } diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 4a38978..7ccbf4c 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -2,6 +2,7 @@ #include #include #include +#include #include #include #include @@ -375,6 +376,45 @@ static int sched_plugin_init(xce_scheduler_t* sched) return 0; } +static int sched_reconnect_cb(xcecli_ctx_t c, void* arg) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + + (void)c; + if (sched == NULL) { + return -EINVAL; + } + + atomic_store_explicit(&sched->needs_rejoin, 1, memory_order_release); + return 0; +} + +int sched_rejoin_if_needed(xce_scheduler_t* sched) +{ + int rc; + + if (sched == NULL) { + return -EINVAL; + } + + if (!atomic_load_explicit(&sched->needs_rejoin, memory_order_acquire)) { + return 0; + } + + atomic_store_explicit(&sched->needs_rejoin, 0, memory_order_release); + INFO("%s: controller reconnected — re-joining as scheduler", __func__); + + rc = xcecli_join(sched->apictx, HostRole_scheduler, NULL); + if (rc < 0) { + WARN("%s: rejoin failed rc=%d; will re-try on the next reconnect", + __func__, rc); + return rc; + } + + INFO("%s: rejoin ok", __func__); + return 0; +} + int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, const xce_cluster_config_t* cluster_cfg, xce_request_handler_t handler) @@ -466,6 +506,7 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, } xcecli_set_handler(sched->apictx, handler, sched); + xcecli_set_reconnect_handler(sched->apictx, sched_reconnect_cb, sched); /* HA mode: iterate candidates until one accepts as primary */ if (sched->cluster_cfg.candidates_count > 0) { @@ -495,6 +536,8 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, continue; } xcecli_set_handler(sched->apictx, handler, sched); + xcecli_set_reconnect_handler(sched->apictx, sched_reconnect_cb, + sched); rc = xcecli_start(sched->apictx); if (rc < 0) { @@ -822,21 +865,56 @@ int sched_update(xce_scheduler_t* sched) sched->pend_updates[1] = q; mylock_unlock(&(sched->lock)); + /* Apply any queued config update FIRST so job/worker/queue updates + * always see the queues/resources they reference (join AND rejoin re-push + * ordering, bug-1513). The config and dataset pushes are separate wire + * messages dispatched to different vcpu threads, so their enqueue order + * is not guaranteed — a job update processed before its config would find + * NULL j->queue and crash. Park non-config items on a staging queue. */ + dataq_t staging; + dataq_init(&staging, NULL); + while (1) { data_item_t* it; xce_sched_update_request_t* req; - int rc; it = dataq_deq(q, NULL); if (it == NULL) { break; } + req = ITPTR_TO_UPDATE_REQ(it); + if (req->op != XCE_SCHED_OP_UPDATE_CONFIG) { + dataq_enq(&staging, it); + continue; + } + + int rc = sched_init_config(sched, req->u.config); + if (rc >= 0) { + /* ownership passed to sched->config */ + req->u.config = NULL; + } + sched_pend_update_killer(it); + if (rc < 0) { + WARN("%s: config update failed: %s (%d)", __func__, + strerror(-rc), rc); + } + } + + while (1) { + data_item_t* it; + xce_sched_update_request_t* req; + int rc; + + it = dataq_deq(&staging, NULL); + if (it == NULL) { + break; + } + req = ITPTR_TO_UPDATE_REQ(it); switch (req->op) { case XCE_SCHED_OP_UPDATE_CONFIG: - /* this will only be handled when scheduler has been started, so - * should never get here */ + /* handled in the pre-pass above; cannot reach here */ rc = -EINVAL; break; case XCE_SCHED_OP_UPDATE_JOB: @@ -957,6 +1035,8 @@ int sched_update(xce_scheduler_t* sched) } } + dataq_fini(&staging); + return 0; }