Skip to content

fix: persist daemon watches across client disconnects #1833

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
LeoLan-Digital wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: persist daemon watches across client disconnects #1833

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 48 additions & 17 deletions src/daemon/application.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
}
}
Expand Down Expand Up @@ -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;
}
}
Expand Down Expand Up @@ -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);
}
Expand Down
102 changes: 102 additions & 0 deletions tests/test_daemon_application.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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);
Expand Down
Loading