fix: persist daemon watches across client disconnects

Signed-off-by: LeoLan Digital <deine-email@example.com>

Author: LeoLan Digital

src/daemon/application.c

@@ -149,11 +149,11 @@ struct cbm_daemon_application_job {
     cbm_daemon_application_job_t *next;
 };
 
-/* A watcher-triggered physical job is owned by the exact live sessions that
- * currently subscribe to its project/root watch. The callback waiting for the
- * job is only a storage waiter; it is deliberately not an ownership
- * subscription, so the worker is cancelled when the last matching session
- * disconnects even while unrelated daemon sessions remain alive. */
+/* A watcher-triggered physical job is normally owned by the exact live
+ * sessions subscribed to its project/root watch. In permanent daemon mode the
+ * watcher itself additionally owns the physical job so background indexing can
+ * continue while no client session is connected. The callback remains a
+ * storage waiter in both modes and releases daemon ownership on completion. */
 struct cbm_daemon_application_watch_job_subscription {
     cbm_daemon_application_session_t *session;
     cbm_daemon_application_job_t *job;
@@ -424,7 +424,19 @@ static void application_release_session_watch_locked(cbm_daemon_application_sess
     if (watch->subscribers > 0) {
         watch->subscribers--;
     }
-    if (watch->subscribers == 0) {
+
+    /*
+     * A permanent daemon owns the physical project watch independently of
+     * individual client sessions. Session disconnect releases only that
+     * session's logical subscription; the watcher remains available for
+     * background change detection while the daemon generation is alive.
+     *
+     * Non-permanent applications retain the historical session-scoped
+     * lifecycle and remove the physical watch after the final subscriber.
+     * Project deletion/pruning and daemon shutdown remove persistent watches
+     * through their dedicated lifecycle paths.
+     */
+    if (!session->application->permanent && watch->subscribers == 0) {
         application_remove_watch_locked(session->application, watch);
     }
 }
@@ -3375,27 +3387,42 @@ static int application_background_index(cbm_daemon_application_t *application,
     cbm_mutex_lock(&application->mutex);
     cbm_daemon_application_watch_t *watch =
         require_live_watch ? application_find_watch_locked(application, project_name) : NULL;
-    bool watch_live = !require_live_watch ||
-                      (watch && watch->subscribers > 0 && strcmp(watch->root, canonical_root) == 0);
+    bool watch_live =
+        !require_live_watch ||
+        (watch && strcmp(watch->root, canonical_root) == 0 &&
+         (watch->subscribers > 0 || application->permanent));
     size_t watch_owner_count = 0;
     bool watch_subscriptions_ok = true;
+    bool watcher_owns_job = false;
     cbm_daemon_application_job_t *job =
         watch_live ? application_job_subscribe_locked(application, project_key, canonical_root,
                                                       args, &subscribe_status)
                    : NULL;
     if (job && require_live_watch) {
         watch_subscriptions_ok = application_watch_job_subscribe_sessions_locked(
             application, watch, job, &watch_owner_count);
-        if (watch_subscriptions_ok && watch_owner_count > 0) {
-            job->watcher_waiters++;
-        } else if (!watch_subscriptions_ok) {
+
+        if (!watch_subscriptions_ok) {
             subscribe_status = APPLICATION_JOB_SUBSCRIBE_ALLOCATION_FAILED;
-        }
-        /* application_job_subscribe_locked() lends the caller one ordinary
-         * subscriber. A watcher callback is only a storage waiter: exact live
-         * session subscriptions above own the physical work. */
-        application_job_unsubscribe_locked(job);
-        if (!watch_subscriptions_ok || watch_owner_count == 0) {
+            application_job_unsubscribe_locked(job);
+            job = NULL;
+        } else if (application->permanent) {
+            /*
+             * The ordinary subscription acquired above is retained as daemon
+             * ownership. This allows the physical watcher to run indexing even
+             * while no client session currently owns the project watch.
+             */
+            watcher_owns_job = true;
+            job->watcher_waiters++;
+        } else if (watch_owner_count > 0) {
+            /*
+             * Historical non-permanent behavior: exact live sessions own the
+             * physical work; the watcher callback itself is only a waiter.
+             */
+            job->watcher_waiters++;
+            application_job_unsubscribe_locked(job);
+        } else {
+            application_job_unsubscribe_locked(job);
             job = NULL;
         }
     }
@@ -3428,6 +3455,10 @@ static int application_background_index(cbm_daemon_application_t *application,
                 if (job->watcher_waiters > 0) {
                     job->watcher_waiters--;
                 }
+                if (watcher_owns_job) {
+                    application_job_unsubscribe_locked(job);
+                    watcher_owns_job = false;
+                }
             } else {
                 application_job_unsubscribe_locked(job);
             }

tests/test_daemon_application.c

@@ -3621,6 +3621,107 @@ TEST(daemon_application_watcher_job_follows_exact_live_watch_owners) {
     PASS();
 }
 
+
+TEST(daemon_application_permanent_watch_survives_disconnect_and_indexes_without_session) {
+    app_watch_race_fixture_t fixture;
+    bool fixture_ready = app_watch_race_fixture_init(&fixture, 47);
+
+    if (fixture_ready) {
+        cbm_daemon_application_set_permanent(fixture.application, true);
+    }
+
+    int watch_before_disconnect =
+        fixture_ready ? cbm_watcher_watch_count(fixture.watcher) : -1;
+
+    if (fixture.session) {
+        fixture.callbacks.session_cancel(fixture.callbacks.context, fixture.session);
+    }
+
+    int watch_after_cancel =
+        fixture_ready ? cbm_watcher_watch_count(fixture.watcher) : -1;
+
+    if (fixture.session) {
+        fixture.callbacks.session_close(fixture.callbacks.context, fixture.session);
+        fixture.session = NULL;
+    }
+
+    int watch_after_close =
+        fixture_ready ? cbm_watcher_watch_count(fixture.watcher) : -1;
+
+    /*
+     * No client session remains at this point. A permanent daemon-owned
+     * physical watch must nevertheless be allowed to launch its background
+     * indexing job.
+     */
+    app_watcher_index_thread_t request = {
+        .application = fixture.application,
+        .project = fixture.project,
+        .root = fixture.root,
+        .pause_before_subscribe = false,
+        .result = -1,
+    };
+    atomic_init(&request.ready, false);
+    atomic_init(&request.proceed, true);
+    atomic_init(&request.done, false);
+
+    cbm_thread_t thread;
+    bool thread_started =
+        fixture_ready &&
+        watch_after_close == 1 &&
+        cbm_thread_create(&thread, 0, app_watcher_index_thread, &request) == 0;
+
+    bool worker_started =
+        thread_started && app_wait_for_atomic_int(&fixture.fake.starts, 1);
+
+    int watch_during_job =
+        fixture_ready ? cbm_watcher_watch_count(fixture.watcher) : -1;
+
+    /*
+     * The fake worker deliberately waits until the test releases it.
+     * Let the daemon-owned watcher job finish normally.
+     */
+    if (worker_started) {
+        atomic_store(&fixture.fake.allow_completion, true);
+    }
+
+    bool request_done =
+        thread_started && app_wait_for_atomic_bool(&request.done, true);
+
+    bool thread_joined =
+        request_done && cbm_thread_join(&thread) == 0;
+
+    int watch_after_job =
+        fixture_ready ? cbm_watcher_watch_count(fixture.watcher) : -1;
+
+    int destroys_before_cleanup =
+        fixture_ready ? atomic_load(&fixture.fake.destroys) : -1;
+
+    bool cleaned = app_watch_race_fixture_finish(&fixture);
+
+    ASSERT_TRUE(fixture_ready);
+    ASSERT_EQ(watch_before_disconnect, 1);
+
+    /* Permanent mode: disconnect must not unwatch the project. */
+    ASSERT_EQ(watch_after_cancel, 1);
+    ASSERT_EQ(watch_after_close, 1);
+
+    /* No session exists, yet the watcher still owns background indexing. */
+    ASSERT_TRUE(thread_started);
+    ASSERT_TRUE(worker_started);
+    ASSERT_EQ(watch_during_job, 1);
+
+    ASSERT_TRUE(request_done);
+    ASSERT_TRUE(thread_joined);
+    ASSERT_EQ(request.result, 0);
+    ASSERT_EQ(destroys_before_cleanup, 1);
+
+    /* Completing the background job must not consume the persistent watch. */
+    ASSERT_EQ(watch_after_job, 1);
+
+    ASSERT_TRUE(cleaned);
+    PASS();
+}
+
 TEST(daemon_application_late_watcher_session_owns_active_watcher_job) {
     app_watch_race_fixture_t fixture;
     bool fixture_ready = app_watch_race_fixture_init(&fixture, 45);
@@ -5116,6 +5217,7 @@ SUITE(daemon_application) {
     RUN_TEST(daemon_application_stale_watcher_callback_is_rejected_at_job_admission);
     RUN_TEST(daemon_application_final_cancel_drains_admitted_watcher_job);
     RUN_TEST(daemon_application_watcher_job_follows_exact_live_watch_owners);
+    RUN_TEST(daemon_application_permanent_watch_survives_disconnect_and_indexes_without_session);
     RUN_TEST(daemon_application_late_watcher_session_owns_active_watcher_job);
     RUN_TEST(daemon_application_serializes_adr_mutation_with_index_job);
     RUN_TEST(daemon_application_reserved_mutation_delays_worker_start);