feat: drive GC from sched_end_session — live_jobs mark, seq_cst fence, gc_sweep (6-8, 6-12)

src/scheduler/src/scheduler.c

@@ -760,6 +760,9 @@ static int sched_account_jobs(xce_scheduler_t* sched)
 
 void sched_begin_session(xce_scheduler_t* sched)
 {
+    atomic_store_explicit(&sched->session_active, 1, memory_order_release);
+    atomic_store_explicit(&sched->staging_merge_complete, 0, memory_order_release);
+
     plugin_begin_session(sched);
 
     sched_reset_shared_resval(sched);
@@ -772,9 +775,101 @@ void sched_begin_session(xce_scheduler_t* sched)
 
 void sched_end_session(xce_scheduler_t* sched)
 {
+    atomic_store_explicit(&sched->session_active, 0, memory_order_release);
+
     plugin_end_session(sched);
 
-    return;
+    /* Guard: GC requires the evaluation pipeline to be idle */
+    if (sched_eval_in_progress(sched) || !sched_staging_merge_complete(sched)) {
+        TRACE("%s: GC skipped — eval_in_progress=%d staging_merge_complete=%d",
+              __func__,
+              sched_eval_in_progress(sched),
+              sched_staging_merge_complete(sched));
+        return;
+    }
+
+    /* Collect live jobs into a flat array for the GC mark phase */
+    iterator_t        it;
+    xce_sched_job_t*  job       = NULL;
+    int               njobs     = 0;
+    int               cap       = 64;
+    xce_sched_job_t** live_jobs = calloc(cap, sizeof(*live_jobs));
+
+    if (live_jobs != NULL) {
+        int rc = iter_make(&it, sched->jobs);
+        if (rc >= 0) {
+            for (;;) {
+                rc = iter_next(&it, (void**)&job);
+                if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
+                    break;
+                }
+                if (njobs >= cap) {
+                    int               newcap = cap * 2;
+                    xce_sched_job_t** tmp    = realloc(live_jobs,
+                                                       newcap * sizeof(*live_jobs));
+                    if (tmp == NULL) {
+                        break;
+                    }
+                    live_jobs = tmp;
+                    cap       = newcap;
+                }
+                live_jobs[njobs++] = job;
+            }
+            if (rc == RTTI_ITER_NO_DATA) {
+                rc = 0;
+            }
+            iter_destroy(&it);
+        }
+    }
+
+    /* Mark phase */
+    gc_clear_marks();
+    for (int i = 0; i < njobs; i++) {
+        xce_resreq_t* rr = live_jobs[i]->resreq;
+        if (rr == NULL) {
+            continue;
+        }
+        if (rr->tag == XCE_RESREQ_PRIMARY) {
+            xce_primary_resreq_t* pr = rr->u.primary;
+            if (pr && pr->select) {
+                gc_mark_select_tree(pr->select->root);
+            }
+        } else if (rr->tag == XCE_RESREQ_COMPOUND) {
+            xce_compound_resreq_t* cp = rr->u.compound;
+            if (cp != NULL) {
+                for (int j = 0; j < cp->primaries.count; j++) {
+                    xce_primary_resreq_t* pr = xlist_get(&cp->primaries, j);
+                    if (pr && pr->select) {
+                        gc_mark_select_tree(pr->select->root);
+                    }
+                }
+            }
+        } else if (rr->tag == XCE_RESREQ_ALTER) {
+            xce_alter_resreq_t* alt = rr->u.alter;
+            if (alt != NULL) {
+                for (int j = 0; j < alt->compounds.count; j++) {
+                    xce_compound_resreq_t* cp = xlist_get(&alt->compounds, j);
+                    if (cp == NULL) {
+                        continue;
+                    }
+                    for (int k = 0; k < cp->primaries.count; k++) {
+                        xce_primary_resreq_t* pr = xlist_get(&cp->primaries, k);
+                        if (pr && pr->select) {
+                            gc_mark_select_tree(pr->select->root);
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    /* Fence: mark writes visible before sweep reads them */
+    atomic_thread_fence(memory_order_seq_cst);
+
+    gc_sweep_select_nodes(NULL, 0);
+    free(live_jobs);
+
+    atomic_store_explicit(&sched->staging_merge_complete, 1, memory_order_release);
 }
 
 int sched_add_job(xce_scheduler_t* sched, xce_sched_job_t* j)

src/scheduler/testsuite/test.c

@@ -196,6 +196,30 @@ test_meet_host_selection_always_false_fails(void)
     sel_optimize_test_reset();
 }
 
+static void
+test_gc_predicates_idle_invariant(void)
+{
+    xce_scheduler_t sched;
+    memset(&sched, 0, sizeof(sched));
+    /* staging_merge_complete=1 means GC is safe to run */
+    atomic_store(&sched.staging_merge_complete, 1);
+
+    sel_optimize_set(1);
+    sel_node_table_init();
+
+    /* Verify the atomic fields are readable — testing the predicate contract */
+    TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.session_active));
+    TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.eval_in_progress));
+    TEST_ASSERT_EQUAL_INT(1, atomic_load(&sched.staging_merge_complete));
+
+    /* GC operations on an empty node table must not crash */
+    gc_clear_marks();
+    gc_sweep_select_nodes(NULL, 0);
+
+    sel_node_table_fini();
+    sel_optimize_test_reset();
+}
+
 #define TEST(x)                                                                \
     do {                                                                       \
         if (check_test(&arguments, #x)) {                                      \
@@ -215,6 +239,7 @@ int main(int argc, char* argv[])
     TEST(test_meet_host_selection_null_select_passes);
     TEST(test_meet_host_selection_always_true_passes);
     TEST(test_meet_host_selection_always_false_fails);
+    TEST(test_gc_predicates_idle_invariant);
 
     return UNITY_END();
 }