diff --git a/src/daemon/application.c b/src/daemon/application.c index 7e4a35be7..ce0cfdaf8 100644 --- a/src/daemon/application.c +++ b/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,10 +3387,13 @@ 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) @@ -3386,16 +3401,28 @@ static int application_background_index(cbm_daemon_application_t *application, 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); } diff --git a/tests/test_daemon_application.c b/tests/test_daemon_application.c index c639ace2f..b0b55e4f6 100644 --- a/tests/test_daemon_application.c +++ b/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);