Function application_job_subscribe_locked

src/daemon/application.c:1582–1643

Caller holds application->mutex. Keeping watcher ownership validation and * this admission in the same critical section closes the unwatch race. */

/* Caller holds application->mutex. Keeping watcher ownership validation and
 * this admission in the same critical section closes the unwatch race. */
static cbm_daemon_application_job_t *application_job_subscribe_locked(
    cbm_daemon_application_t *application, const char *project_key, const char *root_path,
    const char *args_json, application_job_subscribe_status_t *status_out) {
    *status_out = APPLICATION_JOB_SUBSCRIBE_UNAVAILABLE;
    if (application->stopping) {
        return NULL;
    }
    cbm_daemon_application_job_t *job =
        application_find_active_job_locked(application, project_key);
    if (job) {
        if (job->cancel_requested) {
            *status_out = APPLICATION_JOB_SUBSCRIBE_CANCELLING;
            return NULL;
        }
        if (!application_index_args_equal(job->args_json, args_json)) {
            *status_out = APPLICATION_JOB_SUBSCRIBE_OPTIONS_CONFLICT;
            return NULL;
        }
        job->subscribers++;
        *status_out = APPLICATION_JOB_SUBSCRIBE_OK;
        return job;
    }

    if (application_active_job_count_locked(application) >= application->physical_job_limit) {
        char limit[32];
        (void)snprintf(limit, sizeof(limit), "%zu", application->physical_job_limit);
        cbm_log_warn("daemon.index.admission_busy", "limit", limit, "project", project_key);
        *status_out = APPLICATION_JOB_SUBSCRIBE_BUSY;
        return NULL;
    }

    job = calloc(1, sizeof(*job));
    if (job) {
        job->project_key = strdup(project_key);
        job->root_path = strdup(root_path);
        job->args_json = strdup(args_json);
    }
    if (!job || !job->project_key || !job->root_path || !job->args_json) {
        application_job_free(job);
        *status_out = APPLICATION_JOB_SUBSCRIBE_ALLOCATION_FAILED;
        return NULL;
    }
    job->application = application;
    job->subscribers = 1;
    job->next = application->jobs;
    application->jobs = job;
    if (application_job_thread_create(&job->thread, job) == 0) {
        job->thread_started = true;
    } else {
        /* The job was linked only so a concurrently started thread could
         * observe its reservation. No thread exists on this path, so roll the
         * reservation back synchronously and let background callers retry. */
        application->jobs = job->next;
        job->next = NULL;
        application_job_free(job);
        *status_out = APPLICATION_JOB_SUBSCRIBE_UNAVAILABLE;
        cbm_log_warn("daemon.index.thread_start_failed", "action", "retry");
        return NULL;