feat(scheduler-controller-msg-gap): store job rusage on the job + free leaked op payloads (tasks 5.5, 5.8; bug-765 G-5/G-3)

5.5 (Req 6 / G-5): the JOB_RUSAGE notify path called the plugin but never
stored the usage data — xce_sched_job_t had no rusage field (UB-6.1 unmet).
- Add xce_job_rusages_t* rusages to xce_sched_job_t.
- New sched_apply_job_rusage() stores the report ON the job (ownership
  transferred), replacing any prior via free_JobRusages_ptr; -ENOENT unknown,
  -EINVAL NULL. Dequeue NULLs the request pointer on success; kill_sched_job
  releases job->rusages.

5.8 (G-3): sched_pend_update_killer left five new-op payloads on empty
break cases — a per-message leak since ownership is transferred on enqueue.
- Free each with the exact generated deep-freer: free_QueueConf_ptr,
  free_Resource_ptr[], free_ResourceVal_ptr[], free_ResValList_ptr,
  free_JobRusages_ptr. Safe: handlers nullify on enqueue, makeResource copies
  via makeInstance, dequeue copies values (no aliasing).

Tests: sched_job_update_test 7/7 (3 update + 4 rusage), regression
sched_test 15/15, ASAN clean (no bad-free/double-free).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/scheduler/include/sched/types.h

@@ -93,6 +93,10 @@ typedef struct xce_sched_job_ {
     uint32_t             maxtasks;
     xce_job_alloc_t*     allocs;
     xce_job_account_t*   accounts;
+    /* Latest resource-usage report from the controller (Req 6 / bug-765 G-5).
+     * Owned by the job: replaced on each JOB_RUSAGE update, freed in
+     * kill_sched_job. NULL until the first rusage report. */
+    xce_job_rusages_t*   rusages;
 } xce_sched_job_t;
 
 typedef struct {

src/scheduler/include/scheduler.h

@@ -139,6 +139,7 @@ extern int                  sched_add_job(xce_scheduler_t*, xce_sched_job_t*);
 extern int sched_add_queue(xce_scheduler_t*, xce_sched_queue_t*);
 extern int sched_update_existing_queue(xce_scheduler_t*, xce_queue_t*);
 extern int sched_update_existing_job(xce_scheduler_t*, xce_job_t*);
+extern int sched_apply_job_rusage(xce_scheduler_t*, xce_job_rusages_t*);
 extern int sched_add_user(xce_scheduler_t*, xce_sched_user_t*);
 extern int sched_add_worker(xce_scheduler_t*, xce_sched_worker_t*);
 extern int sched_add_project(xce_scheduler_t*, xce_sched_project_t*);

src/scheduler/src/schedobj.c

@@ -16,9 +16,15 @@
 
 void kill_sched_job(void* arg)
 {
-    if (arg == NULL) {
+    xce_sched_job_t* job = (xce_sched_job_t*)arg;
+
+    if (job == NULL) {
         return;
     }
+
+    /* release the stored rusage report, if any (Req 6 / bug-765 G-5).
+     * free_JobRusages_ptr is NULL-safe and clears the field. */
+    free_JobRusages_ptr(&job->rusages);
 }
 
 request_res_t* make_resource_request(char* res, float value)
@@ -584,3 +590,38 @@ int sched_update_existing_job(xce_scheduler_t* sched, xce_job_t* updated)
 
     return 0;
 }
+
+/*
+ * sched_apply_job_rusage — store a controller rusage report ON the matching
+ * job (Req 6, UB-6.1/6.2/6.3; bug-765 G-5). On success the job TAKES OWNERSHIP
+ * of `rusages` (the caller must not free it); the previous report, if any, is
+ * released. Returns:
+ *   -EINVAL  rusages is NULL
+ *   -ENOENT  no job with rusages->jobid (no state modified, ownership retained
+ *            by the caller so it can free the payload)
+ *        0  stored on the job and plugins notified
+ */
+int sched_apply_job_rusage(xce_scheduler_t* sched, xce_job_rusages_t* rusages)
+{
+    xce_sched_job_t* job;
+
+    if (rusages == NULL) {
+        return -EINVAL;
+    }
+
+    job = sched_get_job(sched, rusages->jobid);
+    if (job == NULL) {
+        return -ENOENT;
+    }
+
+    /* replace any prior report; the job now owns this one. free_JobRusages_ptr
+     * is NULL-safe and clears the field before we reassign. */
+    if (job->rusages != rusages) {
+        free_JobRusages_ptr(&job->rusages);
+    }
+    job->rusages = rusages;
+
+    plugin_job_rusage_update(sched, job, rusages);
+
+    return 0;
+}

src/scheduler/src/scheduler.c

@@ -311,15 +311,33 @@ static void sched_pend_update_killer(data_item_t* it)
         break;
     case XCE_SCHED_OP_JSTATE_CHANGE:
         break;
+    /* The five notify/dataset payloads below are owned by the request (the
+     * handler/dataset path nullifies its own pointer on enqueue). The dequeue
+     * copies values / makeResource constructs a fresh instance, so none of
+     * these are aliased into live state — free them here (bug-765 G-3). */
     case XCE_SCHED_OP_UPDATE_QUEUE:
+        free_QueueConf_ptr(&(req->u.queueupdate.queue));
         break;
     case XCE_SCHED_OP_UPDATE_RESOURCES:
+        for (i = 0; i < req->u.resourceupdate.nresources; i++) {
+            free_Resource_ptr(&(req->u.resourceupdate.resources[i]));
+        }
+        free(req->u.resourceupdate.resources);
         break;
     case XCE_SCHED_OP_UPDATE_RESVALS_FULL:
+        for (i = 0; i < req->u.resvalfullupdate.nvals; i++) {
+            free_ResourceVal_ptr(&(req->u.resvalfullupdate.vals[i]));
+        }
+        free(req->u.resvalfullupdate.vals);
         break;
     case XCE_SCHED_OP_UPDATE_RESVALS:
+        free_ResValList_ptr(&(req->u.resvalupdate.resvals));
         break;
     case XCE_SCHED_OP_JOB_RUSAGE:
+        /* NULL after a successful sched_apply_job_rusage (job took ownership);
+         * non-NULL on the -ENOENT/-EINVAL path, freed here (bug-765 G-5).
+         * free_JobRusages_ptr is NULL-safe. */
+        free_JobRusages_ptr(&(req->u.job_rusage.rusages));
         break;
     default:
         break;
@@ -820,18 +838,16 @@ int sched_update(xce_scheduler_t* sched)
             rc = 0;
             break;
         }
-        case XCE_SCHED_OP_JOB_RUSAGE: {
-            xce_sched_job_t*   job    = NULL;
-            xce_job_rusages_t* rusages = req->u.job_rusage.rusages;
-            job = sched_get_job(sched, rusages->jobid);
-            if (job == NULL) {
-                rc = -ENOENT;
-            } else {
-                plugin_job_rusage_update(sched, job, rusages);
-                rc = 0;
+        case XCE_SCHED_OP_JOB_RUSAGE:
+            /* Store the rusage report ON the job (Req 6 / bug-765 G-5). On
+             * success the job takes ownership; clear the request pointer so the
+             * killer does not free it. On failure (-ENOENT/-EINVAL) ownership
+             * stays with the request and the killer frees it. */
+            rc = sched_apply_job_rusage(sched, req->u.job_rusage.rusages);
+            if (rc == 0) {
+                req->u.job_rusage.rusages = NULL;
             }
             break;
-        }
         default:
             rc = -EINVAL;
             break;

src/scheduler/testsuite/test_job_update.c

@@ -42,6 +42,10 @@ xce_sched_project_t* sched_get_project(xce_scheduler_t* s, char* n)
 }
 int plugin_job_update(void* s, xce_sched_job_t* j) { (void)s; (void)j; return 0; }
 int plugin_queue_update(void* s, xce_sched_queue_t* q) { (void)s; (void)q; return 0; }
+int plugin_job_rusage_update(void* s, xce_sched_job_t* j, xce_job_rusages_t* r)
+{
+    (void)s; (void)j; (void)r; return 0;
+}
 xce_job_account_t* make_job_account(xce_job_alloc_t* a) { (void)a; return NULL; }
 void free_job_account(xce_job_account_t* a) { (void)a; }
 int sched_add_user(xce_scheduler_t* s, xce_sched_user_t* u) { (void)s; (void)u; return 0; }
@@ -221,6 +225,76 @@ test_update_null_payload_returns_einval(void)
     TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
 }
 
+/* Job rusage storage (Req 6 / bug-765 G-5) */
+
+static xce_job_rusages_t* make_rusages(uint64_t jobid)
+{
+    /* count==0 / usages==NULL keeps free_job_rusages a clean structural free. */
+    xce_job_rusages_t* r = (xce_job_rusages_t*)calloc(1, sizeof(*r));
+    r->jobid = jobid;
+    return r;
+}
+
+static void
+test_rusage_stored_on_existing_job(void)
+{
+    xce_scheduler_t sched;
+    fixture_init(&sched);
+
+    xce_sched_job_t*   job = seed_job(&sched, 42);
+    xce_job_rusages_t* ru  = make_rusages(42);
+
+    int rc = sched_apply_job_rusage(&sched, ru);
+
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    /* stored ON the job (ownership transferred to the job) */
+    TEST_ASSERT_EQUAL_PTR(ru, job->rusages);
+    /* the job object itself is unchanged identity */
+    TEST_ASSERT_EQUAL_PTR(job, sched_get_job(&sched, 42));
+    /* job owns ru now; kill_sched_job/free_job_rusages will release it */
+}
+
+static void
+test_rusage_replace_frees_prior(void)
+{
+    xce_scheduler_t sched;
+    fixture_init(&sched);
+
+    xce_sched_job_t*   job = seed_job(&sched, 7);
+    xce_job_rusages_t* ru1 = make_rusages(7);
+    xce_job_rusages_t* ru2 = make_rusages(7);
+
+    TEST_ASSERT_EQUAL_INT(0, sched_apply_job_rusage(&sched, ru1));
+    TEST_ASSERT_EQUAL_PTR(ru1, job->rusages);
+    /* second report replaces the first (prior freed, no leak/crash) */
+    TEST_ASSERT_EQUAL_INT(0, sched_apply_job_rusage(&sched, ru2));
+    TEST_ASSERT_EQUAL_PTR(ru2, job->rusages);
+}
+
+static void
+test_rusage_unknown_job_returns_enoent(void)
+{
+    xce_scheduler_t sched;
+    fixture_init(&sched);
+
+    xce_job_rusages_t* ru = make_rusages(999); /* no such job */
+
+    int rc = sched_apply_job_rusage(&sched, ru);
+
+    TEST_ASSERT_EQUAL_INT(-ENOENT, rc);
+    /* on -ENOENT the caller retains ownership */
+    free_JobRusages_ptr(&ru);
+}
+
+static void
+test_rusage_null_returns_einval(void)
+{
+    xce_scheduler_t sched;
+    fixture_init(&sched);
+
+    TEST_ASSERT_EQUAL_INT(-EINVAL, sched_apply_job_rusage(&sched, NULL));
+}
+
 int main(int argc, char* argv[])
 {
     (void)argc; (void)argv;
@@ -228,5 +302,9 @@ int main(int argc, char* argv[])
     RUN_TEST(test_update_existing_job_mutates_in_place, 0);
     RUN_TEST(test_update_unknown_job_returns_enoent_no_create, 0);
     RUN_TEST(test_update_null_payload_returns_einval, 0);
+    RUN_TEST(test_rusage_stored_on_existing_job, 0);
+    RUN_TEST(test_rusage_replace_frees_prior, 0);
+    RUN_TEST(test_rusage_unknown_job_returns_enoent, 0);
+    RUN_TEST(test_rusage_null_returns_einval, 0);
     return UNITY_END();
 }