lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(worker): parse job file transfers under the apiserver's "transfer_files" key (bug-1937)

Browse source 
Cause: the apiserver writes file transfers to exdata as "transfer_files"
(controllers/jobs.go processJobRequest), but parse_exdata() only matched
"transfer_file", so the transfers were logged as an unknown key and
dropped.

Fix: parse_exdata() accepts "transfer_files" and still the older
"transfer_file"; if both are present the first list wins instead of
leaking it. New worker unit test test_exdata_parse.c parses the exdata
JSON the apiserver produces (transfers, env, interactive, limits).
Stage-in/out itself is still not implemented in the worker.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-10 08:37:47 +00:00
parent d01ced9bd0
commit b9747ad2ff
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
3 changed files with 177 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

28
src/worker/src/xcecd_exdata.c
View file

@ -12,7 +12,7 @@
/**
* @brief Parses the JSON string, populates an environment variable linked list,
* and populates TransferInfo from transfer_file.
* and populates TransferInfo from transfer_files.
*
* @param exdata The JSON string to parse.
* @param env_vars Pointer to the head pointer of the env_var_t linked list.
@ -111,20 +111,29 @@ bool parse_exdata(const char* exdata, env_var_t** env_vars,
*env_vars = new_node;
}
}
// --- Handle 'transfer_file' ---
else if (strcmp(key, "transfer_file") == 0) {
// --- Handle 'transfer_files' ---
/* bug-1937: the apiserver writes "transfer_files"; "transfer_file"
* is the older spelling, still accepted. */
else if (strcmp(key, "transfer_files") == 0
|| strcmp(key, "transfer_file") == 0) {
if (!cJSON_IsArray(item)) {
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"Error: 'transfer_file' is not a JSON array.");
"Error: '%s' is not a JSON array.", key);
success = false;
continue;
}
if (transfer_infos->ops != NULL) {
XCECD_WORKER_LOG(XCE_LOG_WARN,
"Ignoring '%s': transfers already parsed.",
key);
continue;
}
int array_size = cJSON_GetArraySize(item);
if (array_size <= 0) {
XCECD_WORKER_LOG(XCE_LOG_DEBUG3,
"'transfer_file' array is empty.");
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "'%s' array is empty.", key);
continue;
}
@ -144,10 +153,9 @@ bool parse_exdata(const char* exdata, env_var_t** env_vars,
for (int i = 0; i < array_size; ++i) {
cJSON* transfer_item = cJSON_GetArrayItem(item, i);
if (!cJSON_IsObject(transfer_item)) {
XCECD_WORKER_LOG(
XCE_LOG_ERROR,
"Error: Item %d in 'transfer_file' is not an object.",
i);
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"Error: Item %d in '%s' is not an object.",
i, key);
success = false; // Mark as failure but continue parsing
// other items

13
src/worker/testsuite/meson.build
View file

@ -234,6 +234,19 @@ test('worker exdata_jobindex test', exdata_jobindex_test,
timeout: 60,
env: ['ASAN_OPTIONS=detect_leaks=0'])
# bug-1937: parse_exdata()/parse_limits() on the exdata JSON the apiserver
# builds ("transfer_files" key) -- white-box #include of xcecd_exdata.c.
exdata_parse_test = executable('exdata_parse_test',
'test_exdata_parse.c',
include_directories: [worker_test_incdir],
dependencies: worker_test_deps,
install: false)
test('worker exdata_parse test', exdata_parse_test,
suite: 'worker',
timeout: 60,
env: ['ASAN_OPTIONS=detect_leaks=0'])
# bug-1908: pending-start signal table (pending_signal.c).
pending_signal_test = executable('pending_signal_test',
['test_pending_signal.c',

146
src/worker/testsuite/test_exdata_parse.c Normal file
View file

@ -0,0 +1,146 @@
/*
* test_exdata_parse.c -- parse_exdata()/parse_limits() on the exdata JSON
* the apiserver builds for a submitted job (bug-1937, bug-1936).
*
* The apiserver (controllers/jobs.go processJobRequest) writes file
* transfers under "transfer_files"; parse_exdata() must read them from that
* key (and still from the older "transfer_file"), together with the env,
* interactive mode and limits in the same object.
*
* White-box #include of ../src/xcecd_exdata.c (same convention as
* test_exdata_jobindex.c). XCECD_WORKER_LOG is redefined to a sink before
* the include: xcelog.h is include-guarded, so the production file's own
* #include "xcelog.h" does not restore the original macro.
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "xcelog.h"
static void ignore_log(int level, const char* fmt, ...)
{
(void)level;
(void)fmt;
}
#undef XCECD_WORKER_LOG
#define XCECD_WORKER_LOG(level, format, ...) \
ignore_log((int)(level), format __VA_OPT__(, ) __VA_ARGS__)
#include "../src/xcecd_exdata.c"
/* What json.Marshal of the apiserver's extendDataMap gives (keys sorted)
* for a job submitted with an env var, -W 1, -M 1G, interactive TTY and
* two file transfers. */
static const char* APISERVER_EXDATA
= "{\"interactive\":1,"
"\"limits\":{\"rss_limit\":1024,\"run_limit\":60},"
"\"sub_env_vars\":{\"A\":\"1\"},"
"\"transfer_files\":["
"{\"local_file\":\"/tmp/in.dat\",\"op\":\">\",\"remote_file\":\"\"},"
"{\"local_file\":\"out.dat\",\"op\":\"<\","
"\"remote_file\":\"/scratch/out.dat\"}]}";
static env_var_t* env;
static transfer_info_t tinfo;
static void free_env(env_var_t* head)
{
while (head) {
env_var_t* next = head->next;
free(head->key);
free(head->value);
free(head);
head = next;
}
}
void setUp(void)
{
env = NULL;
memset(&tinfo, 0, sizeof(tinfo));
}
void tearDown(void)
{
free_env(env);
env = NULL;
free_transfer_info(&tinfo);
}
static void test_apiserver_transfer_files_parsed(void)
{
TEST_ASSERT_TRUE(parse_exdata(APISERVER_EXDATA, &env, &tinfo, 1));
TEST_ASSERT_EQUAL_INT(2, tinfo.count);
TEST_ASSERT_NOT_NULL(tinfo.ops);
TEST_ASSERT_EQUAL_STRING(">", tinfo.ops[0].op);
TEST_ASSERT_EQUAL_STRING("/tmp/in.dat", tinfo.ops[0].local_file);
TEST_ASSERT_EQUAL_STRING("", tinfo.ops[0].remote_file);
TEST_ASSERT_EQUAL_STRING("<", tinfo.ops[1].op);
TEST_ASSERT_EQUAL_STRING("out.dat", tinfo.ops[1].local_file);
TEST_ASSERT_EQUAL_STRING("/scratch/out.dat", tinfo.ops[1].remote_file);
}
static void test_apiserver_env_and_interactive_parsed(void)
{
TEST_ASSERT_TRUE(parse_exdata(APISERVER_EXDATA, &env, &tinfo, 1));
TEST_ASSERT_NOT_NULL(env);
TEST_ASSERT_EQUAL_STRING("A", env->key);
TEST_ASSERT_EQUAL_STRING("1", env->value);
TEST_ASSERT_NULL(env->next);
TEST_ASSERT_EQUAL_INT(TTY, tinfo.interactive_mode);
}
static void test_legacy_transfer_file_key_still_parsed(void)
{
const char* legacy = "{\"transfer_file\":[{\"local_file\":\"a\","
"\"op\":\">\",\"remote_file\":\"b\"}]}";
TEST_ASSERT_TRUE(parse_exdata(legacy, &env, &tinfo, 1));
TEST_ASSERT_EQUAL_INT(1, tinfo.count);
TEST_ASSERT_EQUAL_STRING("a", tinfo.ops[0].local_file);
TEST_ASSERT_EQUAL_STRING("b", tinfo.ops[0].remote_file);
}
static void test_both_keys_keep_first_list(void)
{
const char* both = "{\"transfer_files\":[{\"local_file\":\"a\","
"\"op\":\">\",\"remote_file\":\"\"}],"
"\"transfer_file\":[{\"local_file\":\"x\","
"\"op\":\"<\",\"remote_file\":\"y\"},"
"{\"local_file\":\"z\",\"op\":\"<\","
"\"remote_file\":\"w\"}]}";
TEST_ASSERT_TRUE(parse_exdata(both, &env, &tinfo, 1));
TEST_ASSERT_EQUAL_INT(1, tinfo.count);
TEST_ASSERT_EQUAL_STRING("a", tinfo.ops[0].local_file);
}
static void test_apiserver_limits_parsed(void)
{
res_limits_t limits;
memset(&limits, 0, sizeof(limits));
TEST_ASSERT_TRUE(parse_limits(APISERVER_EXDATA, &limits));
TEST_ASSERT_EQUAL_UINT64(60, limits.run_limit); /* seconds */
TEST_ASSERT_EQUAL_UINT64(1024, limits.memory_max); /* MB */
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_apiserver_transfer_files_parsed);
RUN_TEST(test_apiserver_env_and_interactive_parsed);
RUN_TEST(test_legacy_transfer_file_key_still_parsed);
RUN_TEST(test_both_keys_keep_first_list);
RUN_TEST(test_apiserver_limits_parsed);
return UNITY_END();
}