From e7d717d5cabdb2fc6641eb8ff19cb57979db6056 Mon Sep 17 00:00:00 2001 From: Matt Davis Date: Sat, 31 Oct 2026 09:57:32 -0400 Subject: [PATCH] Fix deadlock when health worker startup partially fails Keep workers at a startup gate until all threads exist. On a later pthread_create failure, gated workers can exit and be joined without waiting for the lifecycle lock held by startup. Preserve same-runtime retries and concurrent start/shutdown ownership. Nine real-thread fault/ownership cases plus four existing type controls pass under ASan/UBSan on main. The new suite reproduced five failures before the fix. Register the regression in the existing native CI job. --- .github/workflows/integration-test.yml | 3 +- src/telemetry/system_health.c | 18 ++ tests/unit/cmake/system_health/startup.cmake | 4 + tests/unit/test_system_health_startup.c | 267 +++++++++++++++++++ 4 files changed, 291 insertions(+), 1 deletion(-) create mode 100644 tests/unit/cmake/system_health/startup.cmake create mode 100644 tests/unit/test_system_health_startup.c diff --git a/.github/workflows/integration-test.yml b/.github/workflows/integration-test.yml index f4f825289..c779d15ec 100644 --- a/.github/workflows/integration-test.yml +++ b/.github/workflows/integration-test.yml @@ -99,6 +99,7 @@ jobs: test_memory \ test_request_response \ test_shutdown_coordinator \ + test_system_health_startup \ test_detection_config \ test_detection_model_motion \ test_db_streams \ @@ -145,7 +146,7 @@ jobs: run: | cd build ctest --output-on-failure -V \ - -R "test_storage_pressure$|test_storage_pressure_extended|test_detection_result_structures|test_stream_frame_rate|test_recording_video_params|test_mp4_segment_boundaries|test_storage_retention_sqlite|test_storage_manager_retention|test_storage_target_pressure_cleanup|test_storage_migration|test_storage_archive_s3|test_storage_archive_operator|test_authorization|test_api_handlers_recordings_playback|test_db_storage_targets|test_db_storage_policies|test_api_handlers_storage_targets|test_api_handlers_storage_policies|test_config|test_logger|test_strings|test_memory|test_request_response|test_shutdown_coordinator|test_detection_config|test_detection_model_motion|test_db_streams|test_db_camera_favorites|test_api_handlers_camera_favorites|test_db_backup|test_mp4_failed_open_memory|test_db_recordings_extended|test_db_detections|test_db_zones|test_db_events|test_db_auth|test_db_transactions|test_db_maintenance|test_db_query_builder|test_logger_json|test_batch_delete_progress|test_db_recordings_sync|test_httpd_utils|test_zone_filter|test_onvif_soap_fault|test_faac_encoder|test_audio_transcoder|test_mp4_audio_recovery|test_mp4_forced_reconnect|test_mp4_recovery_watchdog|test_audio_voice_enhancement|test_audio_voice_filter_faults|test_udt_recording|test_detection_current_frame|test_detection_io_recovery|test_hls_cleanup|test_stream_manager|test_stream_state|test_packet_buffer|test_timestamp_manager|test_motion_trigger_parse|test_motion_max_area|test_url_utils" + -R "test_storage_pressure$|test_storage_pressure_extended|test_detection_result_structures|test_stream_frame_rate|test_recording_video_params|test_mp4_segment_boundaries|test_storage_retention_sqlite|test_storage_manager_retention|test_storage_target_pressure_cleanup|test_storage_migration|test_storage_archive_s3|test_storage_archive_operator|test_authorization|test_api_handlers_recordings_playback|test_db_storage_targets|test_db_storage_policies|test_api_handlers_storage_targets|test_api_handlers_storage_policies|test_config|test_logger|test_strings|test_memory|test_request_response|test_shutdown_coordinator|test_detection_config|test_detection_model_motion|test_db_streams|test_db_camera_favorites|test_api_handlers_camera_favorites|test_db_backup|test_mp4_failed_open_memory|test_db_recordings_extended|test_db_detections|test_db_zones|test_db_events|test_db_auth|test_db_transactions|test_db_maintenance|test_db_query_builder|test_logger_json|test_batch_delete_progress|test_db_recordings_sync|test_httpd_utils|test_zone_filter|test_onvif_soap_fault|test_faac_encoder|test_audio_transcoder|test_mp4_audio_recovery|test_mp4_forced_reconnect|test_mp4_recovery_watchdog|test_audio_voice_enhancement|test_audio_voice_filter_faults|test_udt_recording|test_detection_current_frame|test_detection_io_recovery|test_hls_cleanup|test_stream_manager|test_stream_state|test_packet_buffer|test_timestamp_manager|test_motion_trigger_parse|test_motion_max_area|test_url_utils|test_system_health_startup$" - name: Generate coverage report if: always() diff --git a/src/telemetry/system_health.c b/src/telemetry/system_health.c index af991646a..0f3b360f9 100644 --- a/src/telemetry/system_health.c +++ b/src/telemetry/system_health.c @@ -57,6 +57,7 @@ typedef struct { bool initialized; bool started; bool requested; + bool startup_ready; /* Protected by lock; no collection before full startup. */ } tier_worker_t; typedef struct system_health_runtime { @@ -761,6 +762,13 @@ static void add_ms_to_timespec(struct timespec *value, uint32_t milliseconds) { static void *tier_worker_main(void *argument) { tier_worker_t *worker = argument; system_health_runtime_t *state = worker->runtime; + /* Startup owns lifecycle_lock until all workers exist, or failed workers + * have been joined. Do not enter collection (which takes that lock) before + * successful publication; a failed start wakes this gate only to exit. */ + pthread_mutex_lock(&worker->lock); + while (atomic_load(&state->running) && !worker->startup_ready) + pthread_cond_wait(&worker->wake, &worker->lock); + pthread_mutex_unlock(&worker->lock); while (atomic_load(&state->running)) { pthread_mutex_lock(&worker->lock); worker->requested = false; @@ -797,6 +805,9 @@ int system_health_start(void) { size_t started = 0U; for (size_t tier = 0; tier < SYSTEM_HEALTH_TIER_COUNT; ++tier) { tier_worker_t *worker = &state->workers[tier]; + pthread_mutex_lock(&worker->lock); + worker->startup_ready = false; + pthread_mutex_unlock(&worker->lock); if (pthread_create(&worker->thread, NULL, tier_worker_main, worker) != 0) break; worker->started = true; @@ -818,6 +829,13 @@ int system_health_start(void) { pthread_mutex_lock(&state->lock); state->stats.worker_threads = SYSTEM_HEALTH_TIER_COUNT; pthread_mutex_unlock(&state->lock); + for (size_t tier = 0; tier < SYSTEM_HEALTH_TIER_COUNT; ++tier) { + tier_worker_t *worker = &state->workers[tier]; + pthread_mutex_lock(&worker->lock); + worker->startup_ready = true; + pthread_cond_signal(&worker->wake); + pthread_mutex_unlock(&worker->lock); + } pthread_mutex_unlock(&lifecycle_lock); return 0; } diff --git a/tests/unit/cmake/system_health/startup.cmake b/tests/unit/cmake/system_health/startup.cmake new file mode 100644 index 000000000..decdf19dc --- /dev/null +++ b/tests/unit/cmake/system_health/startup.cmake @@ -0,0 +1,4 @@ +add_layer3_test(test_system_health_startup) +target_link_options(test_system_health_startup PRIVATE + -Wl,--wrap=pthread_cond_wait -Wl,--wrap=pthread_mutex_lock -Wl,--wrap=pthread_create -Wl,--wrap=pthread_join) +set_tests_properties(test_system_health_startup PROPERTIES TIMEOUT 30) diff --git a/tests/unit/test_system_health_startup.c b/tests/unit/test_system_health_startup.c new file mode 100644 index 000000000..49fb68a08 --- /dev/null +++ b/tests/unit/test_system_health_startup.c @@ -0,0 +1,267 @@ +#define _POSIX_C_SOURCE 200809L +/* Exercise real worker startup, partial failure, retry and concurrent shutdown. */ +#include "../../src/telemetry/system_health.c" +#include "unity.h" +#include +#include +#include + +typedef struct { void *(*entry)(void *); void *argument; } start_argument_t; +static start_argument_t starts[SYSTEM_HEALTH_TIER_COUNT]; +static int fail_after, event_fd = -1; +static atomic_uint created, joined, joined_mask, bad_join; +static pthread_t owned_threads[2 * SYSTEM_HEALTH_TIER_COUNT]; +static atomic_bool fault_consumed; +static int contender_kind; +static atomic_bool contender_go, contender_at_lock, contender_done; +static int contender_result; +static atomic_uint reached_lock; +static _Thread_local bool is_worker, is_contender, reported_lock; +static _Thread_local tier_worker_t *current_worker; +static pid_t owned_child; +static int owned_pipe = -1; + +static void event(char value) { + if (event_fd >= 0 && write(event_fd, &value, 1) != 1) _exit(90); +} +int __real_pthread_mutex_lock(pthread_mutex_t *); +int __wrap_pthread_mutex_lock(pthread_mutex_t *lock) { + if (is_worker && !reported_lock && lock == &lifecycle_lock) { + reported_lock = true; + event('L'); + atomic_fetch_add(&reached_lock, 1); + } + if (is_contender && lock == &lifecycle_lock && !reported_lock) { + reported_lock = true; event('C'); atomic_store(&contender_at_lock, true); + } + return __real_pthread_mutex_lock(lock); +} +static void *worker_entry(void *opaque) { + start_argument_t *start = opaque; + is_worker = true; current_worker = start->argument; + return start->entry(start->argument); +} +int __real_pthread_cond_wait(pthread_cond_t *, pthread_mutex_t *); +int __wrap_pthread_cond_wait(pthread_cond_t *condition, pthread_mutex_t *lock) { + /* The fixed startup gate replaces the original contested-lock boundary. + * Keep the original lifecycle-lock seam too: unchanged source stays RED. */ + if (is_worker && !reported_lock && current_worker && + condition == ¤t_worker->wake && lock == ¤t_worker->lock) { + reported_lock = true; event('G'); atomic_fetch_add(&reached_lock, 1); + } + return __real_pthread_cond_wait(condition, lock); +} +static void await_boundary(atomic_bool *flag) { + uint64_t deadline = monotonic_ms() + 1000; + while (!atomic_load(flag)) { + if (monotonic_ms() >= deadline) _exit(93); + struct timespec pause = {0, 1000000}; nanosleep(&pause, NULL); + } +} +static void checkpoint(unsigned workers) { + uint64_t deadline = monotonic_ms() + 1000; + while (atomic_load(&reached_lock) < workers) { + if (monotonic_ms() >= deadline) _exit(91); + struct timespec pause = {0, 1000000}; nanosleep(&pause, NULL); + } + if (contender_kind) { + atomic_store(&contender_go, true); + await_boundary(&contender_at_lock); + if (atomic_load(&contender_done)) _exit(94); + } +} +int __real_pthread_create(pthread_t *, const pthread_attr_t *, void *(*)(void *), void *); +int __wrap_pthread_create(pthread_t *thread, const pthread_attr_t *attr, + void *(*entry)(void *), void *argument) { + if (entry != tier_worker_main) return __real_pthread_create(thread, attr, entry, argument); + tier_worker_t *worker = argument; + unsigned tier = (unsigned)worker->tier; + if (tier >= SYSTEM_HEALTH_TIER_COUNT) _exit(92); + if (fail_after >= 0 && tier == (unsigned)fail_after && + !atomic_exchange(&fault_consumed, true)) { + checkpoint(tier); + event('F'); + return EAGAIN; + } + starts[tier] = (start_argument_t){entry, argument}; + int result = __real_pthread_create(thread, attr, worker_entry, &starts[tier]); + if (result == 0) { + unsigned index = atomic_load(&created); + if (index >= 2 * SYSTEM_HEALTH_TIER_COUNT) _exit(95); + owned_threads[index] = *thread; + atomic_store(&created, index + 1); + if (contender_kind && fail_after < 0 && tier == SYSTEM_HEALTH_TIER_COUNT - 1) + checkpoint(SYSTEM_HEALTH_TIER_COUNT); + } + return result; +} +int __real_pthread_join(pthread_t, void **); +int __wrap_pthread_join(pthread_t thread, void **result) { + event('J'); + unsigned bit = 0; + for (unsigned i = atomic_load(&created); i > 0; --i) { + unsigned candidate = 1U << (i - 1); + if (!(atomic_load(&joined_mask) & candidate) && pthread_equal(thread, owned_threads[i - 1])) { + bit = candidate; break; + } + } + int status = __real_pthread_join(thread, result); + if (status == 0 && bit && !(atomic_fetch_or(&joined_mask, bit) & bit)) + atomic_fetch_add(&joined, 1); + else atomic_fetch_add(&bad_join, 1); + return status; +} +static void *contender_entry(void *unused) { + (void)unused; is_contender = true; + await_boundary(&contender_go); + if (contender_kind == 1) contender_result = system_health_start(); + else { system_health_shutdown(); contender_result = 0; } + atomic_store(&contender_done, true); + return NULL; +} + +static int fake_collect(void *state, const system_health_collect_context_t *context, + system_health_observation_sink_t *sink) { + (void)state; (void)context; (void)sink; + return 0; +} +static int child_scenario(int failure, int concurrent) { + system_health_options_t options; + system_health_options_defaults(&options); + options.register_builtin_collectors = false; + strcpy(options.hardware_provider, "disabled"); + for (size_t tier = 0; tier < SYSTEM_HEALTH_TIER_COUNT; ++tier) + options.tier_interval_ms[tier] = 60000; + if (system_health_init(&options) != 0 || system_health_provider_registry_count() != 0) return 10; + for (unsigned tier = 0; tier < SYSTEM_HEALTH_TIER_COUNT; ++tier) { + system_health_collector_t collector = {0}; + snprintf(collector.name, sizeof(collector.name), "fixture.%u", tier); + collector.tier = (system_health_sampling_tier_t)tier; + collector.scope = SYSTEM_HEALTH_SCOPE_HOST; + collector.collect = fake_collect; + collector.interval_seconds = 1; + collector.stale_after_seconds = 3; + if (!system_health_register_collector(&collector)) return 11; + } + fail_after = failure; created = joined = 0; atomic_store(&reached_lock, 0); + atomic_store(&joined_mask, 0); atomic_store(&bad_join, 0); + atomic_store(&fault_consumed, false); + contender_kind = concurrent; + atomic_store(&contender_go, false); atomic_store(&contender_at_lock, false); + atomic_store(&contender_done, false); contender_result = -99; + pthread_t contender; + if (concurrent && __real_pthread_create(&contender, NULL, contender_entry, NULL) != 0) return 16; + int result = system_health_start(); + if (concurrent) { + if (result != (failure >= 0 ? -1 : 0)) return 17; + if (__real_pthread_join(contender, NULL) != 0 || !atomic_load(&contender_done) || + contender_result != 0 || !atomic_load(&contender_at_lock)) return 18; + unsigned expected = (failure >= 0 ? (unsigned)failure : SYSTEM_HEALTH_TIER_COUNT); + if (concurrent == 1) { + if (!runtime || !runtime->started) return 19; + if (failure >= 0) expected += SYSTEM_HEALTH_TIER_COUNT; + system_health_shutdown(); + } + if (runtime || atomic_load(&created) != expected || atomic_load(&joined) != expected || + atomic_load(&bad_join) || atomic_load(&joined_mask) != (1U << expected) - 1) return 20; + return 0; + } + if (failure >= 0) { + if (result != -1 || created != (unsigned)failure || joined != created || + atomic_load(&runtime->running) || runtime->started) return 12; + for (size_t tier = 0; tier < SYSTEM_HEALTH_TIER_COUNT; ++tier) + if (runtime->workers[tier].started) return 13; + /* Retry the same initialized runtime, then require actual signal/join + * cleanup. No reset/corruption of production worker state to help it. */ + fail_after = -1; created = joined = 0; atomic_store(&reached_lock, 0); + atomic_store(&joined_mask, 0); + result = system_health_start(); + } + if (result != 0 || created != SYSTEM_HEALTH_TIER_COUNT) return 14; + system_health_shutdown(); + if (runtime || joined != created || atomic_load(&bad_join) || + atomic_load(&joined_mask) != (1U << atomic_load(&created)) - 1) return 15; + return 0; +} +static void run_owned_child(int failure, int concurrent) { + int pipes[2]; TEST_ASSERT_EQUAL_INT(0, pipe(pipes)); + fflush(NULL); + owned_child = fork(); + if (owned_child == 0) { + close(pipes[0]); event_fd = pipes[1]; + int result = child_scenario(failure, concurrent); + close(event_fd); event_fd = -1; + exit(result); /* Run sanitizer leak checks on normal child completion. */ + } + close(pipes[1]); owned_pipe = pipes[0]; + TEST_ASSERT_GREATER_THAN_INT(0, owned_child); + TEST_ASSERT_NOT_EQUAL(-1, fcntl(owned_pipe, F_SETFL, O_NONBLOCK)); + uint64_t deadline = monotonic_ms() + 3000; + int status = 0; bool timeout = false; + for (;;) { + pid_t result = waitpid(owned_child, &status, WNOHANG); + if (result == owned_child) break; + TEST_ASSERT_EQUAL_INT(0, result); + if (monotonic_ms() >= deadline) { + timeout = true; + /* Still an unreaped owned child: PID cannot have been recycled. */ + TEST_ASSERT_EQUAL_INT(0, kill(owned_child, SIGKILL)); + TEST_ASSERT_EQUAL_INT(owned_child, waitpid(owned_child, &status, 0)); + break; + } + struct timespec pause = {0, 1000000}; nanosleep(&pause, NULL); + } + owned_child = 0; + unsigned locks = 0, gates = 0, faults = 0, joins = 0, contenders = 0; + char bytes[128]; ssize_t count; + while ((count = read(owned_pipe, bytes, sizeof(bytes))) > 0) + for (ssize_t i = 0; i < count; ++i) { + if (bytes[i] == 'L') locks++; + if (bytes[i] == 'G') gates++; + if (bytes[i] == 'C') contenders++; + if (bytes[i] == 'F') faults++; + if (bytes[i] == 'J') joins++; + } + TEST_ASSERT_EQUAL_INT(0, close(owned_pipe)); owned_pipe = -1; + printf("startup failure=%d concurrent=%d: reached_lock=%u reached_gate=%u contender=%u injected=%u joins_entered=%u timeout=%d status=%d\n", + failure, concurrent, locks, gates, contenders, faults, joins, timeout, status); + if (failure >= 0) TEST_ASSERT_EQUAL_UINT(1, faults); + if (failure > 0) { + TEST_ASSERT_GREATER_OR_EQUAL_UINT((unsigned)failure, locks + gates); + TEST_ASSERT_GREATER_OR_EQUAL_UINT(1, joins); + } + if (concurrent) TEST_ASSERT_EQUAL_UINT(1, contenders); + TEST_ASSERT_FALSE_MESSAGE(timeout, "owned child exceeded deadline; containment kill is a failed lifecycle result"); + TEST_ASSERT_TRUE(WIFEXITED(status)); + TEST_ASSERT_EQUAL_INT(0, WEXITSTATUS(status)); +} +void setUp(void) { owned_child = 0; owned_pipe = -1; } +void tearDown(void) { + bool leaked = owned_child > 0 || owned_pipe >= 0; + if (owned_child > 0) { kill(owned_child, SIGKILL); waitpid(owned_child, NULL, 0); owned_child = 0; } + if (owned_pipe >= 0) { close(owned_pipe); owned_pipe = -1; } + TEST_ASSERT_FALSE(leaked); +} +static void test_normal_start_shutdown_joins_every_worker(void) { run_owned_child(-1, 0); } +static void test_first_worker_failure_can_retry_and_shutdown(void) { run_owned_child(0, 0); } +static void test_second_worker_failure_releases_first_and_can_retry(void) { run_owned_child(1, 0); } +static void test_third_worker_failure_releases_prior_workers_and_can_retry(void) { run_owned_child(2, 0); } +static void test_fourth_worker_failure_releases_prior_workers_and_can_retry(void) { run_owned_child(3, 0); } +static void test_partial_failure_serializes_concurrent_retry(void) { run_owned_child(2, 1); } +static void test_partial_failure_serializes_concurrent_shutdown(void) { run_owned_child(2, 2); } +static void test_successful_start_serializes_concurrent_start(void) { run_owned_child(-1, 1); } +static void test_successful_start_serializes_concurrent_shutdown(void) { run_owned_child(-1, 2); } + +int main(void) { + UNITY_BEGIN(); + RUN_TEST(test_normal_start_shutdown_joins_every_worker); + RUN_TEST(test_first_worker_failure_can_retry_and_shutdown); + RUN_TEST(test_second_worker_failure_releases_first_and_can_retry); + RUN_TEST(test_third_worker_failure_releases_prior_workers_and_can_retry); + RUN_TEST(test_fourth_worker_failure_releases_prior_workers_and_can_retry); + RUN_TEST(test_partial_failure_serializes_concurrent_retry); + RUN_TEST(test_partial_failure_serializes_concurrent_shutdown); + RUN_TEST(test_successful_start_serializes_concurrent_start); + RUN_TEST(test_successful_start_serializes_concurrent_shutdown); + return UNITY_END(); +}