feat: add memory-aware global index scheduling

Signed-off-by: Zhiyu <zhiyuzhang001@gmail.com>

Author: zhiyuzhang001-a11y

src/daemon/application.c

@@ -36,6 +36,7 @@ typedef struct {
     cbm_index_worker_poll_t (*poll)(void *context, cbm_daemon_application_worker_t worker,
                                     const cbm_index_worker_result_t **result_out);
     bool (*cancel)(void *context, cbm_daemon_application_worker_t worker);
+    bool (*observed_rss)(void *context, cbm_daemon_application_worker_t worker, size_t *bytes_out);
     const char *(*log_path)(void *context, cbm_daemon_application_worker_t worker);
     void (*destroy)(void *context, cbm_daemon_application_worker_t worker);
 } cbm_daemon_application_worker_ops_t;

src/daemon/application.h

@@ -37,6 +37,9 @@ int cbm_daemon_application_background_initializes_for_test(void);
  * that a request QUEUED rather than erroring or starting. */
 int cbm_daemon_application_busy_queue_waits_for_test(void);
 
+/* Sum of sampled process-group RSS high-water marks for active index jobs. */
+size_t cbm_daemon_application_observed_index_rss_for_test(cbm_daemon_application_t *application);
+
 /* Build the JSON-RPC error substituted for a reply too large to frame (#1375).
  * Exposed because the alternative — driving a real >10 MiB reply — needs a
  * ~20k-node index, a fixture cost the unit suite should not carry. The

src/daemon/application_internal.h

@@ -413,6 +413,8 @@ typedef struct {
     int count;
     int capacity;
     int max_files;
+    size_t max_total_bytes;
+    size_t total_bytes;
     uint64_t deadline_ms;
     bool count_only;
     bool collect_excluded;
@@ -504,8 +506,15 @@ static void fl_add(file_list_t *fl, const char *abs_path, const char *rel_path,
         fl->limit_exceeded = true;
         return;
     }
+    size_t measured_size = size > 0 ? (size_t)size : 0;
+    if (fl->count_only && (fl->total_bytes > fl->max_total_bytes ||
+                           measured_size > fl->max_total_bytes - fl->total_bytes)) {
+        fl->limit_exceeded = true;
+        return;
+    }
     if (fl->count_only) {
         fl->count++;
+        fl->total_bytes += measured_size;
         return;
     }
     if (fl->count >= fl->capacity) {
@@ -1117,7 +1126,8 @@ static cbm_discover_status_t discover_impl(const char *repo_path, const cbm_disc
                                            int *excluded_count_out,
                                            cbm_ignored_file_t **ignored_out, int *ignored_count_out,
                                            int *ignored_total_out, bool count_only, int max_files,
-                                           uint64_t deadline_ms) {
+                                           size_t max_total_bytes, uint64_t deadline_ms,
+                                           size_t *total_bytes_out) {
     if (excluded_out) {
         *excluded_out = NULL;
     }
@@ -1133,6 +1143,9 @@ static cbm_discover_status_t discover_impl(const char *repo_path, const cbm_disc
     if (ignored_total_out) {
         *ignored_total_out = 0;
     }
+    if (total_bytes_out) {
+        *total_bytes_out = 0;
+    }
     if (!repo_path || !out || !count || (count_only && max_files < 0)) {
         return CBM_DISCOVER_ERROR;
     }
@@ -1206,6 +1219,7 @@ static cbm_discover_status_t discover_impl(const char *repo_path, const cbm_disc
     /* Walk */
     file_list_t fl = {
         .max_files = count_only ? max_files : -1,
+        .max_total_bytes = count_only ? max_total_bytes : SIZE_MAX,
         .deadline_ms = count_only ? deadline_ms : 0,
         .count_only = count_only,
         .collect_excluded = !count_only && excluded_out != NULL,
@@ -1224,6 +1238,9 @@ static cbm_discover_status_t discover_impl(const char *repo_path, const cbm_disc
         cbm_discover_free_excluded(fl.excluded, fl.excluded_count);
         cbm_discover_free_ignored(fl.ignored, fl.ignored_count);
         *count = fl.count;
+        if (total_bytes_out) {
+            *total_bytes_out = fl.total_bytes;
+        }
         if (fl.failed) {
             return CBM_DISCOVER_ERROR;
         }
@@ -1269,7 +1286,7 @@ int cbm_discover_ex2(const char *repo_path, const cbm_discover_opts_t *opts, cbm
                      cbm_ignored_file_t **ignored_out, int *ignored_count_out,
                      int *ignored_total_out) {
     return discover_impl(repo_path, opts, out, count, excluded_out, excluded_count_out, ignored_out,
-                         ignored_count_out, ignored_total_out, false, 0, 0);
+                         ignored_count_out, ignored_total_out, false, 0, SIZE_MAX, 0, NULL);
 }
 
 cbm_discover_status_t cbm_discover_count_bounded(const char *repo_path,
@@ -1283,10 +1300,36 @@ cbm_discover_status_t cbm_discover_count_bounded(const char *repo_path,
     }
     cbm_file_info_t *files = NULL;
     int count = 0;
-    cbm_discover_status_t status = discover_impl(repo_path, opts, &files, &count, NULL, NULL, NULL,
-                                                 NULL, NULL, true, max_files, deadline_ms);
+    cbm_discover_status_t status =
+        discover_impl(repo_path, opts, &files, &count, NULL, NULL, NULL, NULL, NULL, true,
+                      max_files, SIZE_MAX, deadline_ms, NULL);
+    cbm_discover_free(files, count);
+    *count_out = status == CBM_DISCOVER_ERROR ? -1 : count;
+    return status;
+}
+
+cbm_discover_status_t cbm_discover_measure_bounded(const char *repo_path,
+                                                   const cbm_discover_opts_t *opts, int max_files,
+                                                   size_t max_total_bytes, uint64_t deadline_ms,
+                                                   int *count_out, size_t *total_bytes_out) {
+    if (count_out) {
+        *count_out = -1;
+    }
+    if (total_bytes_out) {
+        *total_bytes_out = 0;
+    }
+    if (!repo_path || !count_out || !total_bytes_out || max_files < 0) {
+        return CBM_DISCOVER_ERROR;
+    }
+    cbm_file_info_t *files = NULL;
+    int count = 0;
+    size_t total_bytes = 0;
+    cbm_discover_status_t status =
+        discover_impl(repo_path, opts, &files, &count, NULL, NULL, NULL, NULL, NULL, true,
+                      max_files, max_total_bytes, deadline_ms, &total_bytes);
     cbm_discover_free(files, count);
     *count_out = status == CBM_DISCOVER_ERROR ? -1 : count;
+    *total_bytes_out = total_bytes;
     return status;
 }
 

src/discover/discover.c

@@ -164,6 +164,15 @@ cbm_discover_status_t cbm_discover_count_bounded(const char *repo_path,
                                                  const cbm_discover_opts_t *opts, int max_files,
                                                  uint64_t deadline_ms, int *count_out);
 
+/* Count indexable files and their aggregate source bytes without retaining a
+ * file array. Stops when either limit would be exceeded. This is the daemon's
+ * pre-spawn resource classifier, so a large job is identified before it can
+ * consume a daily-worker slot. */
+cbm_discover_status_t cbm_discover_measure_bounded(const char *repo_path,
+                                                   const cbm_discover_opts_t *opts, int max_files,
+                                                   size_t max_total_bytes, uint64_t deadline_ms,
+                                                   int *count_out, size_t *total_bytes_out);
+
 /* Like cbm_discover(), but also reports the directory subtrees that were
  * skipped during the walk (hardcoded ALWAYS_SKIP/FAST_SKIP dirs + gitignore
  * matches), so callers can surface which subtrees were dropped (#411).

src/discover/discover.h

@@ -22,11 +22,17 @@
                        * non-ASCII repo path survives CreateProcess (#423/#20) */
 #include <stdlib.h>   /* free */
 #else
+#ifndef __APPLE__
+#include <dirent.h>
+#endif
 #include <errno.h>
 #include <fcntl.h>
 #include <signal.h>
 #ifdef __APPLE__
+#include <libproc.h>
 #include <spawn.h>
+#include <sys/proc_info.h>
+#include <sys/sysctl.h>
 extern char **environ;
 #endif
 #include <sys/stat.h>
@@ -423,6 +429,7 @@ struct cbm_subprocess {
     bool root_reaped;
     uint64_t force_started_ms;
     bool containment_failed;
+    size_t observed_tree_rss;
     cbm_proc_result_t result;
 
 #ifdef _WIN32
@@ -436,6 +443,116 @@ struct cbm_subprocess {
 #endif
 };
 
+static bool cbm_subprocess_tree_rss_sample(cbm_subprocess_t *process, size_t *bytes_out) {
+    *bytes_out = 0;
+#ifdef _WIN32
+    JOBOBJECT_EXTENDED_LIMIT_INFORMATION information;
+    memset(&information, 0, sizeof(information));
+    if (!process->job || !QueryInformationJobObject(process->job, JobObjectExtendedLimitInformation,
+                                                    &information, sizeof(information), NULL)) {
+        return false;
+    }
+    *bytes_out = (size_t)information.PeakJobMemoryUsed;
+    return true;
+#elif defined(__APPLE__)
+    int query[4] = {CTL_KERN, KERN_PROC, KERN_PROC_ALL, 0};
+    size_t length = 0;
+    if (sysctl(query, 3, NULL, &length, NULL, 0) != 0 || length == 0) {
+        return false;
+    }
+    struct kinfo_proc *processes = malloc(length);
+    if (!processes || sysctl(query, 3, processes, &length, NULL, 0) != 0) {
+        free(processes);
+        return false;
+    }
+    size_t total = 0;
+    size_t count = length / sizeof(*processes);
+    for (size_t i = 0; i < count; i++) {
+        if (processes[i].kp_eproc.e_pgid != process->pgid) {
+            continue;
+        }
+        struct rusage_info_v2 usage;
+        memset(&usage, 0, sizeof(usage));
+        if (proc_pid_rusage(processes[i].kp_proc.p_pid, RUSAGE_INFO_V2, (rusage_info_t *)&usage) ==
+            0) {
+            size_t resident = (size_t)usage.ri_resident_size;
+            total = resident > SIZE_MAX - total ? SIZE_MAX : total + resident;
+        }
+    }
+    free(processes);
+    *bytes_out = total;
+    return true;
+#else
+    DIR *directory = opendir("/proc");
+    if (!directory) {
+        return false;
+    }
+    long page_size = sysconf(_SC_PAGESIZE);
+    size_t total = 0;
+    struct dirent *entry = NULL;
+    while (page_size > 0 && (entry = readdir(directory)) != NULL) {
+        char *end = NULL;
+        long pid = strtol(entry->d_name, &end, 10);
+        if (pid <= 0 || !end || *end != '\0') {
+            continue;
+        }
+        char path[128];
+        (void)snprintf(path, sizeof(path), "/proc/%ld/stat", pid);
+        FILE *stat_file = fopen(path, "r");
+        char stat_line[4096];
+        if (!stat_file || !fgets(stat_line, sizeof(stat_line), stat_file)) {
+            if (stat_file) {
+                (void)fclose(stat_file);
+            }
+            continue;
+        }
+        (void)fclose(stat_file);
+        char *command_end = strrchr(stat_line, ')');
+        char state = 0;
+        long parent = 0;
+        long group = 0;
+        if (!command_end || sscanf(command_end + 1, " %c %ld %ld", &state, &parent, &group) != 3 ||
+            group != process->pgid) {
+            continue;
+        }
+        (void)snprintf(path, sizeof(path), "/proc/%ld/statm", pid);
+        FILE *memory_file = fopen(path, "r");
+        unsigned long pages = 0;
+        unsigned long resident_pages = 0;
+        bool read = memory_file && fscanf(memory_file, "%lu %lu", &pages, &resident_pages) == 2;
+        if (memory_file) {
+            (void)fclose(memory_file);
+        }
+        if (read) {
+            size_t resident = resident_pages > SIZE_MAX / (size_t)page_size
+                                  ? SIZE_MAX
+                                  : (size_t)resident_pages * (size_t)page_size;
+            total = resident > SIZE_MAX - total ? SIZE_MAX : total + resident;
+        }
+    }
+    (void)closedir(directory);
+    *bytes_out = total;
+    return true;
+#endif
+}
+
+bool cbm_subprocess_observed_tree_rss(cbm_subprocess_t *process, size_t *bytes_out) {
+    if (!bytes_out) {
+        return false;
+    }
+    *bytes_out = 0;
+    if (!process) {
+        return false;
+    }
+    size_t sample = 0;
+    bool sampled = cbm_subprocess_tree_rss_sample(process, &sample);
+    if (sampled && sample > process->observed_tree_rss) {
+        process->observed_tree_rss = sample;
+    }
+    *bytes_out = process->observed_tree_rss;
+    return sampled || process->observed_tree_rss > 0;
+}
+
 static void cbm_subprocess_result_init(cbm_proc_result_t *result) {
     result->outcome = CBM_PROC_SPAWN_FAILED;
     result->exit_code = -1;

src/foundation/subprocess.c

@@ -125,6 +125,12 @@ cbm_proc_poll_t cbm_subprocess_poll(cbm_subprocess_t *process, cbm_proc_result_t
  * performs signal delivery/escalation. */
 bool cbm_subprocess_request_cancel(cbm_subprocess_t *process);
 
+/* Return the largest observed resident byte total for the complete contained
+ * process tree. The sample is refreshed on each call while the tree is live;
+ * after exit the cached high-water mark remains available until destroy.
+ * false means the platform sample was unavailable and leaves *bytes_out zero. */
+bool cbm_subprocess_observed_tree_rss(cbm_subprocess_t *process, size_t *bytes_out);
+
 /* Release a terminal handle. This never waits or implicitly cancels; passing a
  * still-running handle violates the API contract. NULL is a no-op. */
 void cbm_subprocess_destroy(cbm_subprocess_t *process);

src/foundation/subprocess.h

@@ -779,6 +779,10 @@ bool cbm_index_worker_request_cancel(cbm_index_worker_handle_t *handle) {
            cbm_subprocess_request_cancel(handle->process);
 }
 
+bool cbm_index_worker_observed_tree_rss(cbm_index_worker_handle_t *handle, size_t *bytes_out) {
+    return handle && cbm_subprocess_observed_tree_rss(handle->process, bytes_out);
+}
+
 const char *cbm_index_worker_response_path(const cbm_index_worker_handle_t *handle) {
     return handle ? handle->response_path : NULL;
 }

src/mcp/index_supervisor.c

@@ -179,6 +179,7 @@ cbm_index_worker_poll_t cbm_index_worker_poll(cbm_index_worker_handle_t *handle,
  * bounded containment failure is explicitly surfaced in the result). The
  * owner must stop concurrent cancellation producers before destroy. */
 bool cbm_index_worker_request_cancel(cbm_index_worker_handle_t *handle);
+bool cbm_index_worker_observed_tree_rss(cbm_index_worker_handle_t *handle, size_t *bytes_out);
 
 /* Borrowed diagnostic paths, stable until destroy. Every start uses securely
  * created unique files, so concurrent jobs in one daemon cannot collide. The