# Design: scheduler-controller-msg-gap > Feature: scheduler-controller-msg-gap > Created: 2026-05-11 > Status: Design Generated ## Architecture Pattern & Boundary Map **Pattern:** Extension of existing message handler infrastructure. The scheduler uses a two-phase processing model: 1. **Enqueue phase** (handler.c, runs on IO thread) — lightweight: validate, allocate request struct, enqueue into `pend_updates[0]` 2. **Dequeue phase** (scheduler.c `sched_update()`, runs on main thread) — heavyweight: mutate scheduler state, call plugin hooks, trigger scheduling sessions All new handlers follow this pattern. No architectural boundary changes needed. **Scope boundary:** Only `src/scheduler/` source files, `src/lib/helper.c`, and `src/include/helper.h` are modified. No controller-side changes. No Cap'n Proto schema changes. ## Technology Stack & Alignment | Aspect | Choice | Rationale | |--------|--------|-----------| | Language | C99 | Matches existing codebase convention | | IPC protocol | Cap'n Proto (existing) | No schema changes — only handler gaps filled | | Build system | Meson (existing) | No build changes needed | | Threading model | Same as existing — handler enqueues, main thread dequeues | Preserves thread safety guarantees | | Logging | Existing `TRACE`/`INFO`/`WARN`/`ERROR` macros | Consistent with codebase | | Plugin hooks | Follow existing `plugin_*_update()` pattern | Consistency with established plugin API | ## Components & Interface Contracts ### Component 1: handler.c — Notify handlers (Req 3, 4, 5, 6) Extend `sched_handle_notify()` to process the four currently-stubbed NotifyData events. #### `sched_handle_sched_update_job` ``` Signature: static int sched_handle_sched_update_job( xce_scheduler_t* sched, xce_job_t* job) Behavior: - If job == NULL, return -EINVAL - Enqueue XCE_SCHED_OP_UPDATE_JOB with job pointer - On calloc failure, return -ENOMEM - Return 0 on success ``` **Requirement traceability:** 3.1, 3.2, 3.3 #### `sched_handle_sched_update_queue` ``` Signature: static int sched_handle_sched_update_queue( xce_scheduler_t* sched, xce_queue_t* queue) Behavior: - If queue == NULL, return -EINVAL - Enqueue XCE_SCHED_OP_UPDATE_QUEUE with queue pointer - On calloc failure, return -ENOMEM - Return 0 on success ``` **Requirement traceability:** 4.1, 4.2, 4.3 #### `sched_handle_sched_update_resvals` ``` Signature: static int sched_handle_sched_update_resvals( xce_scheduler_t* sched, xce_resval_list_t* resvals) Behavior: - If resvals == NULL or count == 0, return 0 - Enqueue XCE_SCHED_OP_UPDATE_RESVALS with resvals pointer - On calloc failure, return -ENOMEM - Return 0 on success ``` **Requirement traceability:** 5.1, 5.2, 5.3 #### `sched_handle_sched_job_rusage` ``` Signature: static int sched_handle_sched_job_rusage( xce_scheduler_t* sched, xce_job_rusages_t* rusages) Behavior: - If rusages == NULL, return -EINVAL - Enqueue XCE_SCHED_OP_JOB_RUSAGE with rusages pointer - On calloc failure, return -ENOMEM - Return 0 on success ``` **Requirement traceability:** 6.1, 6.2, 6.3 #### `sched_handle_notify` switch changes The existing switch in `sched_handle_notify()` gains four new cases: | NotifyData event | Handler function | Previous behavior | |---|---|---| | `NotifyData_data_schedUpdateJob` | `sched_handle_sched_update_job(sched, notify->data.sched_update_job)` | TRACE log only | | `NotifyData_data_schedUpdateQueue` | `sched_handle_sched_update_queue(sched, notify->data.sched_update_queue)` | Empty `break` | | `NotifyData_data_schedUpdateResvals` | `sched_handle_sched_update_resvals(sched, notify->data.sched_update_resvals)` | Empty `break` | | `NotifyData_data_schedJobRusages` | `sched_handle_sched_job_rusage(sched, notify->data.sched_job_rusages)` | Empty `break` | ### Component 2: handler.c — Dataset handlers (Req 1) #### `sched_handle_resources` ``` Signature: static int sched_handle_resources( xce_scheduler_t* sched, xce_resource_data_t* resdata) Behavior: - If resdata == NULL or resdata->count == 0, return 0 - Enqueue XCE_SCHED_OP_UPDATE_RESOURCES with resdata pointer - On calloc failure, return -ENOMEM - Return 0 on success ``` **Requirement traceability:** 1.1, 1.3, 1.4 #### `sched_handle_resvals` ``` Signature: static int sched_handle_resvals( xce_scheduler_t* sched, xce_resval_data_t* resvaldata) Behavior: - If resvaldata == NULL or resvaldata->count == 0, return 0 - Enqueue XCE_SCHED_OP_UPDATE_RESVALS_FULL with resvaldata pointer - On calloc failure, return -ENOMEM - Return 0 on success ``` **Requirement traceability:** 1.2, 1.3, 1.4 #### `sched_handle_dataset` switch changes | Data table | Handler function | Previous behavior | |---|---|---| | `Data_data_resources` | `sched_handle_resources(sched, dataset->data.resources)` | Empty `break` | | `Data_data_resvals` | `sched_handle_resvals(sched, dataset->data.resvals)` | Empty `break` | ### Component 3: scheduler.h — New op types and update request fields Extend `xce_sched_op_t` and `xce_sched_update_request_t`: ``` typedef enum { XCE_SCHED_OP_NONE = 0, XCE_SCHED_OP_UPDATE_CONFIG, XCE_SCHED_OP_UPDATE_JOB, XCE_SCHED_OP_JSTATE_CHANGE, XCE_SCHED_OP_UPDATE_WORKER, XCE_SCHED_OP_UPDATE_QUEUE, // NEW XCE_SCHED_OP_UPDATE_RESOURCES, // NEW XCE_SCHED_OP_UPDATE_RESVALS_FULL, // NEW — from dataset bulk resvals XCE_SCHED_OP_UPDATE_RESVALS, // NEW — from notify incremental resvals XCE_SCHED_OP_JOB_RUSAGE // NEW } xce_sched_op_t; ``` Extend `xce_sched_update_request_t` union: ``` struct { xce_queue_t* queue; // for UPDATE_QUEUE } queueupdate; struct { xce_resource_data_t* resdata; // for UPDATE_RESOURCES } resourceupdate; struct { xce_resval_data_t* resvaldata; // for UPDATE_RESVALS_FULL } resvalfullupdate; struct { xce_resval_list_t* resvals; // for UPDATE_RESVALS (notify) } resvalupdate; struct { xce_job_rusages_t* rusages; // for JOB_RUSAGE } job_rusage; ``` ### Component 4: scheduler.c — Dequeue processing in `sched_update()` Extend the `switch (req->op)` in `sched_update()` with new cases: #### `XCE_SCHED_OP_UPDATE_QUEUE` - Find existing queue by name with `sched_get_queue()` - If found: update mutable fields (status, priority, accept_interval, sched_delay, resreq) - If not found: return `-ENOENT` — queues are created only via the initial `Conf4Sched` config push, never via notify - Call `plugin_queue_update()` **Requirement traceability:** 4.1, 4.2 #### `XCE_SCHED_OP_UPDATE_RESOURCES` - Iterate `resdata->resources[]` - For each: call `makeResource(sched->resmgr, r)` to add or update - NULL entries skipped (1.3) **Requirement traceability:** 1.1 #### `XCE_SCHED_OP_UPDATE_RESVALS_FULL` - Iterate `resvaldata->vals[]` - For each: find resource by name in `sched->resmgr`, update value - Unknown resources skipped with WARN log **Requirement traceability:** 1.2 #### `XCE_SCHED_OP_UPDATE_RESVALS` - Iterate `resvals->vals[]` - For each: find worker by host name via `sched_get_worker()` - If found: update worker's `resvals` zarray at the resource index - Unknown hosts skipped (5.2) **Requirement traceability:** 5.1, 5.2, 5.3 #### `XCE_SCHED_OP_JOB_RUSAGE` - Find job by jobid via `sched_get_job()` - If not found: return `-ENOENT` (6.2) - Store rusage data on job object - Call `plugin_job_rusage_update()` **Requirement traceability:** 6.1, 6.2 ### Component 5: scheduler.c — Job update-in-place (Req 3) #### New function: `sched_update_existing_job` ``` Signature: static int sched_update_existing_job( xce_scheduler_t* sched, xce_job_t* updated) Behavior: - Find existing job by jobid via sched_get_job() - If not found: return -ENOENT — jobs are created only via the Data_data_schedjobs dataset push, never via notify - If found: update mutable fields (state, priority, allocs) - Call plugin_job_update() - Return 0 on success, negative on error ``` **Requirement traceability:** 3.1, 3.2, 3.3 This is wired into the existing `XCE_SCHED_OP_UPDATE_JOB` case — no new op type needed since `sched_on_jobs()` already creates new jobs. The change is to make `sched_update_jobs()` also handle job updates for existing jobids by replacing rather than just adding. ### Component 6: plugin.h / plugin.c — New rusage hook (Req 6) Add to `plugin_hooks_t`: ``` typedef int (*plugin_job_rusage_hook_t)(void*, xce_sched_job_t*, xce_job_rusages_t*); ``` Add to `plugin_hooks_t` struct: ``` plugin_job_rusage_hook_t rusage_hook; ``` Add global function: ``` int plugin_job_rusage_update(void* sched, xce_sched_job_t* job, xce_job_rusages_t* rusages); ``` This follows the exact pattern of `plugin_job_state_change()` — iterate registered plugins, call their hook if non-NULL. ### Component 7: handler.c — Dead code removal (Req 2) Remove from `scheduler_req_handler()`: - The entire `Request_body_pushmsg` case block (lines 185–198) - The static functions `sched_handle_jobpush()` and `sched_handle_workerpush()` These are unreachable because `vtask_entry()` in `src/lib/xcectx.c` rewrites all `pushmsg` into `datamsg`. **Requirement traceability:** 2.1, 2.2 ### Component 8: Name-lookup functions (Req 8) > **Layering correction (bug-765 G-7, 2026-06-18):** these two helpers live in DIFFERENT layers. `xce_notify_event_name(int)` is lib-level and lives in `src/lib/helper.c` / `src/include/helper.h`. The scheduler-op name lookup operates on `xce_sched_op_t`, a **scheduler-only** type (`scheduler.h`); `src/lib` must not include scheduler headers (no layering inversion), so it is a **static `sched_op_name(xce_sched_op_t)` in `src/scheduler/src/scheduler.c`**, NOT `xce_sched_op_name` in lib/helper. The original "both in helper.c" text below is superseded for the scheduler-op helper. #### New function: `xce_notify_event_name` (lib — `src/lib/helper.c`) ``` Signature: char* xce_notify_event_name(int event) Behavior: Switch onNotifyData_data_* enum values, return symbolic string. Fallback: "UNKNOWN(%d)" with numeric substitution. ``` Matches pattern of `xce_dataset_table()` — takes int, returns `char*`. #### Scheduler-op name lookup: static `sched_op_name` (scheduler — `src/scheduler/src/scheduler.c`) ``` Signature: static char* sched_op_name(xce_sched_op_t op) Behavior: Switch on XCE_SCHED_OP_* enum values, return symbolic string. Fallback: "UNKNOWN(%d)" with numeric substitution. ``` Kept in the scheduler layer (NOT lib/helper as `xce_sched_op_name`) because `xce_sched_op_t` is a scheduler-only type and `src/lib` must not depend on scheduler headers — see the layering correction above. Required for Requirement 7's error logging in `sched_update()`. #### Changes to `handler.c` logging | Current | Changed to | |---|---| | `TRACE("%s table = %d", __func__, dataset->table)` | `TRACE("%s table = %s", __func__, xce_dataset_table(dataset->table))` | | `TRACE("%s...", __func__)` (in stub handlers) | `INFO("%s event = %s jobid = %s", __func__, xce_notify_event_name(notify->event), jobidString)` | **Requirement traceability:** 8.1, 8.2, 8.3, 8.4 ### Component 9: scheduler.c — Error logging in `sched_update()` (Req 7) In the `sched_update()` main loop, after the `switch (req->op)` block, when `rc < 0`: ``` WARN("%s: op %s failed: %s (%d)", __func__, sched_op_name(req->op), strerror(-rc), rc); ``` **Requirement traceability:** 7.1 ### Component 10: schedobj.c — Queue update function (Req 4) #### New function: `sched_update_existing_queue` ``` Signature: int sched_update_existing_queue( xce_scheduler_t* sched, xce_queue_t* updated) Behavior: - Find existing queue by name via sched_get_queue() - If not found: return -ENOENT — queues are created only via the initial Conf4Sched config push, never via notify - If found: update mutable fields: sched_queue->status = updated->status sched_queue->priority = updated->priority sched_queue->accept_interval = updated->accept_interval sched_queue->sched_delay = updated->sched_delay // resreq is a string field — reallocate and replace - Call plugin_queue_update() - Return 0 on success, negative on error ``` **Requirement traceability:** 4.1, 4.2 ### Component 11: `sched_pend_update_killer` updates (scheduler.c) Extend the `sched_pend_update_killer` destructor to handle new op types: ``` case XCE_SCHED_OP_UPDATE_QUEUE: // queue pointer belongs to scheduler, no free needed break; case XCE_SCHED_OP_UPDATE_RESOURCES: // resource_data rows belong to scheduler, no free needed break; case XCE_SCHED_OP_UPDATE_RESVALS_FULL: // resval_data rows owned by capnp decode, freed by free_Request break; case XCE_SCHED_OP_UPDATE_RESVALS: // resval_list owned by capnp decode, freed by free_Request break; case XCE_SCHED_OP_JOB_RUSAGE: // rusage struct owned by capnp decode, freed by free_Request break; ``` **Important:** Memory ownership must be traced carefully. For `XCE_SCHED_OP_UPDATE_JOB` and `XCE_SCHED_OP_UPDATE_WORKER`, the existing pattern nullifies `dataset->nrows` and `dataset->rows` after enqueue to transfer ownership. New queue/rusage/resvals handlers should follow the same pattern — the request struct owns the data, and `free_Request()` in `xcectx.c` handles cleanup. ## Data Flow Diagrams ### Notify message processing flow ``` Controller (sched.c) Scheduler (handler.c) Scheduler (scheduler.c) ───────────────────── ───────────────────── ───────────────────── xce_notify_req() │ ├── schedStartJob ──────────► sched_job_state_change() ──► enq_pend_update(JSTATE_CHANGE) │ │ ├── schedExitJob ───────────► sched_job_state_change() ──► │ │ │ ├── schedFailJob ───────────► sched_job_state_change() ──► │ │ │ ├── schedUpdateJob ────────► sched_handle_sched_update_job() ─► enq_pend_update(UPDATE_JOB) [NEW] │ │ ├── schedUpdateQueue ──────► sched_handle_sched_update_queue() ─► enq_pend_update(UPDATE_QUEUE) [NEW] │ │ ├── schedUpdateResvals ────► sched_handle_sched_update_resvals() ─► enq_pend_update(UPDATE_RESVALS) [NEW] │ │ └── schedJobRusages ───────► sched_handle_sched_job_rusage() ──► enq_pend_update(JOB_RUSAGE) [NEW] │ ▼ sched_update() main loop ├── UPDATE_JOB ──► sched_update_existing_job() ├── UPDATE_QUEUE ──► sched_update_existing_queue() ├── UPDATE_RESVALS ──► update worker resvals ├── UPDATE_RESOURCES ──► update resmgr [NEW] ├── UPDATE_RESVALS_FULL ──► update sharedInst [NEW] └── JOB_RUSAGE ──► store on job + plugin hook [NEW] ``` ### Dataset message processing flow ``` Controller (dataex.c) Scheduler (handler.c) Scheduler (scheduler.c) ──────────────────── ───────────────────── ───────────────────── xce_push_data_req() │ ├── Data_data_schedjobs ──► sched_on_jobs() ──────────► enq_pend_update(UPDATE_JOB) │ │ ├── Data_data_schedworkers ► sched_on_workers() ───────► enq_pend_update(UPDATE_WORKER) │ │ ├── Data_data_resources ──► sched_handle_resources() ─► enq_pend_update(UPDATE_RESOURCES) [NEW] │ │ └── Data_data_resvals ────► sched_handle_resvals() ───► enq_pend_update(UPDATE_RESVALS_FULL) [NEW] │ ▼ sched_update() main loop ``` ## File Change Summary | File | Change Type | Requirements | |------|-------------|-------------| | `src/scheduler/src/handler.c` | Modified — add handlers, remove dead code, update logging | 1, 2, 3, 4, 5, 6, 8 | | `src/scheduler/include/scheduler.h` | Modified — extend enums and structs | 1, 3, 4, 5, 6 | | `src/scheduler/src/scheduler.c` | Modified — add dequeue cases, error logging, update functions | 1, 3, 4, 5, 6, 7 | | `src/scheduler/include/plugin.h` | Modified — add rusage hook typedef and global function | 6 | | `src/scheduler/src/plugin.c` | Modified — add `plugin_job_rusage_update()` implementation | 6 | | `src/scheduler/src/schedobj.c` | Modified — add `sched_update_existing_queue()` | 4 | | `src/lib/helper.c` | Modified — add `xce_notify_event_name()` (lib-level, int arg); the scheduler-op name lookup is a static `sched_op_name()` in `scheduler.c` instead — see Component 8 layering note | 8 | | `src/include/helper.h` | Modified — declare `xce_notify_event_name()` | 8 | | `src/scheduler/testsuite/test.c` | Modified — add unit tests for all new handlers | All | ## Build Order 1. `src/include/helper.h` — add declarations (no dependencies) 2. `src/lib/helper.c` — add implementations (depends on step 1) 3. `src/scheduler/include/scheduler.h` — extend enums and structs (depends on typemap headers) 4. `src/scheduler/include/plugin.h` — add rusage hook typedef (depends on xce/types.h) 5. `src/scheduler/src/plugin.c` — add `plugin_job_rusage_update()` (depends on step 4) 6. `src/scheduler/src/schedobj.c` — add `sched_update_existing_queue()` (depends on step 3) 7. `src/scheduler/src/scheduler.c` — add dequeue processing, error logging, update functions (depends on steps 3–6) 8. `src/scheduler/src/handler.c` — add handlers, remove dead code, update logging (depends on steps 1–7) 9. `src/scheduler/testsuite/test.c` — add unit tests (depends on all above) ## Risks & Mitigations | Risk | Severity | Mitigation | |------|----------|-----------| | Memory ownership of queued data — capnp-decoded structs freed before dequeue processes them | High | Follow existing pattern: for dataset paths, nullify `dataset->nrows`/`dataset->rows` after enqueue to transfer ownership. For notify paths, copy essential data into the request struct rather than holding pointers to capnp memory. | | `sched_update_existing_job` must handle concurrent modification — handler thread enqueues while main thread processes | Medium | Already mitigated by existing double-buffer swap in `sched_update()`. The handler thread only touches `pend_updates[0]` while the main thread drains `pend_updates[1]`. | | New `UPDATE_QUEUE` op requires queue name lookup — potential lock contention | Low | `sched_get_queue()` already uses read lock. Queue updates are infrequent. | | `sched_op_name` (static, scheduler.c) must handle future enum additions | Low | Follow `xce_dataset_table` pattern — return `"UNKNOWN(%d)"` for unrecognized values. |