From fd9b01ad7ad1ff6bef463d8beb009043cbab5854 Mon Sep 17 00:00:00 2001 From: Pcristin Date: Sun, 11 Oct 2026 22:33:05 +0000 Subject: [PATCH] fix(daemon): wait out Windows local teardown contention Signed-off-by: Pcristin --- src/daemon/ipc.c | 81 ++++++++++------- src/daemon/ipc.h | 8 +- src/main.c | 24 ++--- tests/test_daemon_ipc.c | 90 ++++++++++++++++++- tests/windows/test_daemon_stability.py | 120 +++++++++++++++++++++++++ 5 files changed, 276 insertions(+), 47 deletions(-) diff --git a/src/daemon/ipc.c b/src/daemon/ipc.c index a0c1ddbcdc..2afe499763 100644 --- a/src/daemon/ipc.c +++ b/src/daemon/ipc.c @@ -31,6 +31,8 @@ enum { CBM_DAEMON_IPC_RETRY_INTERVAL_MS = 10, CBM_DAEMON_IPC_COORDINATION_CLEANUP_MS = 500, CBM_DAEMON_IPC_WINDOWS_RENDEZVOUS_RETRY_MS = 250, + CBM_DAEMON_IPC_RELEASE_ERROR = -1, + CBM_DAEMON_IPC_RELEASE_COMPLETE = 1, }; /* Last private-namespace validation refusal, set at the exact failing check. @@ -3988,19 +3990,19 @@ bool cbm_daemon_ipc_local_transition_begin_work(cbm_daemon_ipc_local_transition_ return true; } -bool cbm_daemon_ipc_local_transition_release(cbm_daemon_ipc_local_transition_t **transition_io) { +int cbm_daemon_ipc_local_transition_try_release(cbm_daemon_ipc_local_transition_t **transition_io) { if (!transition_io) { - return false; + return CBM_DAEMON_IPC_RELEASE_ERROR; } cbm_daemon_ipc_local_transition_t *transition = *transition_io; if (!transition) { - return true; + return CBM_DAEMON_IPC_RELEASE_COMPLETE; } if (transition->legacy_fd < 0 || !transition->legacy_process_entry || transition->owner_pid != getpid() || (!transition->work_begun && (transition->startup_v2_fd < 0 || !transition->startup_v2_process_entry))) { - return false; + return CBM_DAEMON_IPC_RELEASE_ERROR; } posix_named_lock_release(transition->legacy_fd, transition->legacy_process_entry); if (!transition->work_begun) { @@ -4008,7 +4010,7 @@ bool cbm_daemon_ipc_local_transition_release(cbm_daemon_ipc_local_transition_t * } free(transition); *transition_io = NULL; - return true; + return CBM_DAEMON_IPC_RELEASE_COMPLETE; } #else /* _WIN32 */ @@ -6203,35 +6205,39 @@ static void win_participant_group_release_complete(cbm_private_file_lock_t **gro } } -static bool win_participant_state_release(const cbm_daemon_ipc_endpoint_t *endpoint, - cbm_private_lock_directory_t *directory, - cbm_private_file_lock_t **startup_v2_io, - win_legacy_mutex_guard_t **legacy_guard_io, - cbm_private_file_lock_t **group_io, HANDLE *sentinel_io) { +static int win_participant_state_release(const cbm_daemon_ipc_endpoint_t *endpoint, + cbm_private_lock_directory_t *directory, + cbm_private_file_lock_t **startup_v2_io, + win_legacy_mutex_guard_t **legacy_guard_io, + cbm_private_file_lock_t **group_io, HANDLE *sentinel_io) { if (!endpoint || !directory || !startup_v2_io || !legacy_guard_io || !group_io || !sentinel_io) { - return false; + return CBM_DAEMON_IPC_RELEASE_ERROR; } if (!*startup_v2_io) { int startup = win_startup_v2_try_acquire(directory, startup_v2_io); if (startup != 1) { - return false; + return startup; } } if (!*legacy_guard_io) { int legacy = win_legacy_mutex_guard_try_acquire(endpoint, legacy_guard_io); if (legacy != 1) { - (void)cbm_private_file_lock_release(startup_v2_io); - return false; + if (cbm_private_file_lock_release(startup_v2_io) != CBM_PRIVATE_FILE_LOCK_OK) { + return CBM_DAEMON_IPC_RELEASE_ERROR; + } + return legacy; } } if (!win_participant_group_release_locked(group_io, sentinel_io)) { - return false; + return CBM_DAEMON_IPC_RELEASE_ERROR; } if (!win_legacy_mutex_guard_release(legacy_guard_io)) { - return false; + return CBM_DAEMON_IPC_RELEASE_ERROR; } - return cbm_private_file_lock_release(startup_v2_io) == CBM_PRIVATE_FILE_LOCK_OK; + return cbm_private_file_lock_release(startup_v2_io) == CBM_PRIVATE_FILE_LOCK_OK + ? CBM_DAEMON_IPC_RELEASE_COMPLETE + : CBM_DAEMON_IPC_RELEASE_ERROR; } cbm_daemon_ipc_listener_t *cbm_daemon_ipc_listen_reserved( @@ -7007,9 +7013,9 @@ bool cbm_daemon_ipc_startup_lock_release(cbm_daemon_ipc_startup_lock_t **lock_io if (!lock) { return true; } - if (!win_participant_state_release(lock->endpoint, lock->directory, &lock->startup_v2_lock, - &lock->legacy_guard, &lock->group_lock, - &lock->legacy_sentinel)) { + if (win_participant_state_release(lock->endpoint, lock->directory, &lock->startup_v2_lock, + &lock->legacy_guard, &lock->group_lock, + &lock->legacy_sentinel) != CBM_DAEMON_IPC_RELEASE_COMPLETE) { return false; } cbm_private_lock_directory_close(lock->directory); @@ -7072,9 +7078,10 @@ bool cbm_daemon_ipc_participant_guard_release(cbm_daemon_ipc_participant_guard_t if (!guard) { return true; } - if (!win_participant_state_release( - guard->endpoint, guard->directory, &guard->teardown_startup_v2_lock, - &guard->teardown_legacy_guard, &guard->group_lock, &guard->legacy_sentinel)) { + if (win_participant_state_release(guard->endpoint, guard->directory, + &guard->teardown_startup_v2_lock, + &guard->teardown_legacy_guard, &guard->group_lock, + &guard->legacy_sentinel) != CBM_DAEMON_IPC_RELEASE_COMPLETE) { return false; } cbm_private_lock_directory_close(guard->directory); @@ -7169,39 +7176,45 @@ bool cbm_daemon_ipc_local_transition_begin_work(cbm_daemon_ipc_local_transition_ return true; } -bool cbm_daemon_ipc_local_transition_release(cbm_daemon_ipc_local_transition_t **transition_io) { +int cbm_daemon_ipc_local_transition_try_release(cbm_daemon_ipc_local_transition_t **transition_io) { if (!transition_io) { - return false; + return CBM_DAEMON_IPC_RELEASE_ERROR; } cbm_daemon_ipc_local_transition_t *transition = *transition_io; if (!transition) { - return true; + return CBM_DAEMON_IPC_RELEASE_COMPLETE; } if (!transition->endpoint || !transition->directory) { - return false; + return CBM_DAEMON_IPC_RELEASE_ERROR; } if (transition->group_lock || transition->legacy_sentinel != INVALID_HANDLE_VALUE || transition->teardown_legacy_guard) { - if (!win_participant_state_release(transition->endpoint, transition->directory, - &transition->startup_v2_lock, - &transition->teardown_legacy_guard, - &transition->group_lock, &transition->legacy_sentinel)) { - return false; + int released = win_participant_state_release( + transition->endpoint, transition->directory, &transition->startup_v2_lock, + &transition->teardown_legacy_guard, &transition->group_lock, + &transition->legacy_sentinel); + if (released != CBM_DAEMON_IPC_RELEASE_COMPLETE) { + return released; } } else if (transition->startup_v2_lock && cbm_private_file_lock_release(&transition->startup_v2_lock) != CBM_PRIVATE_FILE_LOCK_OK) { - return false; + return CBM_DAEMON_IPC_RELEASE_ERROR; } cbm_private_lock_directory_close(transition->directory); transition->directory = NULL; free(transition); *transition_io = NULL; - return true; + return CBM_DAEMON_IPC_RELEASE_COMPLETE; } #endif /* _WIN32 */ +bool cbm_daemon_ipc_local_transition_release(cbm_daemon_ipc_local_transition_t **transition_io) { + return cbm_daemon_ipc_local_transition_try_release(transition_io) == + CBM_DAEMON_IPC_RELEASE_COMPLETE; +} + bool cbm_daemon_ipc_send_frame(cbm_daemon_ipc_connection_t *connection, cbm_daemon_frame_type_t type, uint16_t flags, const void *payload, uint32_t length) { diff --git a/src/daemon/ipc.h b/src/daemon/ipc.h index f8195f3d59..4646618f51 100644 --- a/src/daemon/ipc.h +++ b/src/daemon/ipc.h @@ -320,7 +320,13 @@ bool cbm_daemon_ipc_local_transition_begin_work(cbm_daemon_ipc_local_transition_ * Windows teardown reacquires startup-v2 and the legacy mutex, releases group * SH, closes this participant's sentinel last, then releases mutex and v2. * Success consumes and clears *transition_io; a retryable native cleanup - * failure returns false and retains it. */ + * failure or startup-gate contention returns false and retains it. */ bool cbm_daemon_ipc_local_transition_release(cbm_daemon_ipc_local_transition_t **transition_io); +/* Nonblocking release: 1 consumes and clears *transition_io (also when already + * empty), 0 retains it while another startup owns a required gate, and -1 + * retains it on an unsafe/native I/O failure. A NULL transition_io is invalid. + * Callers can give healthy contention its own wait policy without extending + * the deadline for repeated native cleanup failures. */ +int cbm_daemon_ipc_local_transition_try_release(cbm_daemon_ipc_local_transition_t **transition_io); #endif /* CBM_DAEMON_IPC_H */ diff --git a/src/main.c b/src/main.c index c9c67e83ee..0300263e12 100644 --- a/src/main.c +++ b/src/main.c @@ -1622,20 +1622,24 @@ static bool main_project_lock_manager_close(cbm_project_lock_manager_t **manager } static bool main_local_transition_close(cbm_daemon_ipc_local_transition_t **transition) { - uint64_t deadline = main_deadline_after(MAIN_COORDINATION_CLEANUP_MS); + uint64_t ceiling = main_deadline_after(MAIN_STARTUP_CONTENTION_CEILING_MS); + uint64_t error_deadline = main_deadline_after(MAIN_COORDINATION_CLEANUP_MS); while (transition && *transition) { - /* A failed release always RETAINS the transition and is retriable by - * contract: the Windows release transition must briefly try-hold the - * shared startup-v2/legacy gates, so concurrent one-shot teardowns - * legitimately collide and succeed on retry. Success consumes the - * transition; only never-finishing cleanup is a failure, and the - * deadline escalation below owns that. */ - (void)cbm_daemon_ipc_local_transition_release(transition); + /* Windows teardown reacquires the startup-v2/legacy gates before + * releasing its participant claim. A live startup owner can hold them + * beyond the native cleanup deadline. Retain protection and use the + * acquisition backstop for contention; repeated native errors still + * get only the short cleanup deadline. Success consumes the handle. */ + int status = cbm_daemon_ipc_local_transition_try_release(transition); if (*transition) { - if (cbm_now_ms() >= deadline) { + uint64_t now = cbm_now_ms(); + if (now >= ceiling || (status < 0 && now >= error_deadline)) { main_coordination_cleanup_fail_stop("local_transition_cleanup"); } - cbm_usleep(1000); + if (status == 0) { + error_deadline = main_deadline_after(MAIN_COORDINATION_CLEANUP_MS); + } + cbm_usleep(status == 0 ? 10000 : 1000); } } return true; diff --git a/tests/test_daemon_ipc.c b/tests/test_daemon_ipc.c index 35ad5433cf..ac03da3f38 100644 --- a/tests/test_daemon_ipc.c +++ b/tests/test_daemon_ipc.c @@ -2892,6 +2892,91 @@ TEST(daemon_ipc_local_participants_overlap_and_allow_modern_startup) { PASS(); } +TEST(daemon_ipc_windows_local_transition_release_reports_live_gate_contention) { +#ifndef _WIN32 + PASS(); +#else + static const char key[] = "6060717182829393"; + char parent[TEST_PATH_CAP] = {0}; + char canonical_parent[TEST_PATH_CAP] = {0}; + char runtime_dir[TEST_PATH_CAP] = {0}; + char legacy_pipe[CBM_DAEMON_IPC_WINDOWS_NAME_CAP] = {0}; + char legacy_startup[CBM_DAEMON_IPC_WINDOWS_NAME_CAP] = {0}; + cbm_daemon_ipc_endpoint_t *endpoint = NULL; + cbm_daemon_ipc_local_transition_t *transition = NULL; + cbm_daemon_ipc_startup_lock_t *startup = NULL; + + bool parent_ok = ipc_test_parent_new(parent, "local-release-busy") && + ipc_test_full_path(canonical_parent, parent); + bool names_ok = parent_ok && cbm_daemon_ipc_windows_legacy_names(canonical_parent, key, + legacy_pipe, legacy_startup); + if (names_ok) { + endpoint = cbm_daemon_ipc_endpoint_new(key, parent); + } + if (endpoint) { + ipc_test_copy_path(runtime_dir, cbm_daemon_ipc_endpoint_runtime_dir(endpoint)); + } + int acquired = + endpoint ? cbm_daemon_ipc_local_transition_try_acquire(endpoint, &transition) : -1; + int sealed = acquired == 1 ? cbm_daemon_ipc_local_transition_seal_legacy(transition) : -1; + int lifetime = + sealed == 1 ? cbm_daemon_ipc_local_transition_lifetime_probe(endpoint, transition) : -1; + bool begun = lifetime == 0 && cbm_daemon_ipc_local_transition_begin_work(transition); + int startup_acquired = begun ? cbm_daemon_ipc_startup_lock_try_acquire(endpoint, &startup) : -1; + int blocked = cbm_daemon_ipc_local_transition_try_release(&transition); + bool retained_while_busy = transition != NULL; + bool boolean_blocked = cbm_daemon_ipc_local_transition_release(&transition); + /* Keep the already-validated named mutex alive after the startup owner + * exits. Holding just this mutex exercises the second contention path. */ + wchar_t *legacy_name = names_ok ? cbm_utf8_to_wide(legacy_startup) : NULL; + HANDLE legacy_mutex = startup_acquired == 1 && legacy_name + ? OpenMutexW(SYNCHRONIZE | MUTEX_MODIFY_STATE, FALSE, legacy_name) + : NULL; + free(legacy_name); + bool startup_released = cbm_daemon_ipc_startup_lock_release(&startup); + DWORD mutex_wait = + startup_released && legacy_mutex ? WaitForSingleObject(legacy_mutex, 0) : WAIT_FAILED; + bool mutex_owned = mutex_wait == WAIT_OBJECT_0 || mutex_wait == WAIT_ABANDONED; + int legacy_blocked = + mutex_owned ? cbm_daemon_ipc_local_transition_try_release(&transition) : -1; + bool retained_while_legacy_busy = transition != NULL; + bool mutex_released = mutex_owned && ReleaseMutex(legacy_mutex); + if (legacy_mutex) { + (void)CloseHandle(legacy_mutex); + } + int legacy_retained = endpoint ? cbm_daemon_ipc_legacy_generation_probe(endpoint) : -1; + int released = cbm_daemon_ipc_local_transition_try_release(&transition); + bool consumed = transition == NULL; + int legacy_after = endpoint ? cbm_daemon_ipc_legacy_generation_probe(endpoint) : -1; + + (void)cbm_daemon_ipc_startup_lock_release(&startup); + (void)cbm_daemon_ipc_local_transition_release(&transition); + cbm_daemon_ipc_endpoint_free(endpoint); + ipc_test_remove_tree(runtime_dir, parent); + + ASSERT_TRUE(parent_ok); + ASSERT_TRUE(names_ok); + ASSERT_EQ(acquired, 1); + ASSERT_EQ(sealed, 1); + ASSERT_EQ(lifetime, 0); + ASSERT_TRUE(begun); + ASSERT_EQ(startup_acquired, 1); + ASSERT_EQ(blocked, 0); + ASSERT_TRUE(retained_while_busy); + ASSERT_FALSE(boolean_blocked); + ASSERT_TRUE(startup_released); + ASSERT_EQ(mutex_wait, WAIT_OBJECT_0); + ASSERT_EQ(legacy_blocked, 0); + ASSERT_TRUE(retained_while_legacy_busy); + ASSERT_TRUE(mutex_released); + ASSERT_EQ(legacy_retained, 1); + ASSERT_EQ(released, 1); + ASSERT_TRUE(consumed); + ASSERT_EQ(legacy_after, 0); + PASS(); +#endif +} + TEST(daemon_ipc_windows_local_transition_release_retries_retained_mutex) { #ifndef _WIN32 PASS(); @@ -2919,7 +3004,7 @@ TEST(daemon_ipc_windows_local_transition_release_retries_retained_mutex) { if (begun) { cbm_daemon_ipc_windows_legacy_guard_release_failures_set_for_test(1); } - bool first_release = cbm_daemon_ipc_local_transition_release(&transition); + int first_release = cbm_daemon_ipc_local_transition_try_release(&transition); bool retained_after_failure = transition != NULL; bool retry_release = cbm_daemon_ipc_local_transition_release(&transition); cbm_daemon_ipc_windows_legacy_guard_release_failures_set_for_test(0); @@ -2935,7 +3020,7 @@ TEST(daemon_ipc_windows_local_transition_release_retries_retained_mutex) { ASSERT_EQ(sealed, 1); ASSERT_EQ(lifetime, 0); ASSERT_TRUE(begun); - ASSERT_FALSE(first_release); + ASSERT_EQ(first_release, -1); ASSERT_TRUE(retained_after_failure); ASSERT_TRUE(retry_release); ASSERT_NULL(transition); @@ -5792,6 +5877,7 @@ SUITE(daemon_ipc) { RUN_TEST(daemon_ipc_activation_probe_ignores_matching_startup_claim_only); RUN_TEST(daemon_ipc_local_transition_coexists_with_active_daemon_lifetime); RUN_TEST(daemon_ipc_local_participants_overlap_and_allow_modern_startup); + RUN_TEST(daemon_ipc_windows_local_transition_release_reports_live_gate_contention); RUN_TEST(daemon_ipc_windows_local_transition_release_retries_retained_mutex); RUN_TEST(daemon_ipc_windows_startup_release_retains_retry_authority); RUN_TEST(daemon_ipc_frame_timeout_poisons_connection); diff --git a/tests/windows/test_daemon_stability.py b/tests/windows/test_daemon_stability.py index 7bc8381821..ee25e769a0 100644 --- a/tests/windows/test_daemon_stability.py +++ b/tests/windows/test_daemon_stability.py @@ -421,6 +421,125 @@ def section_churn_stability(binary, work): kill_pid(daemon_pid) +def section_local_cleanup_waits_for_startup_gate(binary, work): + """A live startup owner must not turn completed local work into failure.""" + if os.name != "nt": + print("SKIP: local teardown reacquires the startup gate only on Windows") + return True + import msvcrt + + cache = section_dirs(work, "cleanup-contention") + env = cli_env(cache) + env["CBM_LOG_LEVEL"] = "info" + env["CBM_LOG_FORMAT"] = "text" + warm = subprocess.run([binary, "cli", "list_projects"], capture_output=True, + timeout=90, env=env) + if warm.returncode != 0: + raise SetupFailure("could not initialize the private runtime:\n" + out_text(warm)) + gates = [] + for parent, _, files in os.walk(runtime_for(cache)): + if "cbm-startup-v2.lock" in files: + gates.append(os.path.join(parent, "cbm-startup-v2.lock")) + if len(gates) != 1: + raise SetupFailure("expected one private startup gate, got %r" % gates) + + repo = os.path.join(work, "cleanup-repo") + os.makedirs(repo) + for index in range(200): + with open(os.path.join(repo, "unit_%d.py" % index), "w") as handle: + handle.write("".join("def function_%d_%d():\n return %d\n" + % (index, item, item) for item in range(100))) + err_path = os.path.join(work, "cleanup-contention.stderr") + process = None + locked = False + traces = {} + + def read_trace(): + # The daemon supervises indexing in a worker. Worker progress goes to + # its own durable log; the outer CLI only reports the final response. + paths = [err_path] + log_dir = os.path.join(cache, "logs") + if os.path.isdir(log_dir): + paths.extend(os.path.join(log_dir, name) for name in os.listdir(log_dir) + if name.startswith(".worker-log-")) + for path in paths: + try: + with open(path, "rb") as handle: + traces[path] = handle.read() + except FileNotFoundError: + pass # The supervisor removes a successful worker's log. + return b"\n".join(traces.values()) + + with open(gates[0], "r+b") as gate, open(err_path, "wb") as errors: + try: + process = subprocess.Popen( + [binary, "cli", "index_repository", "--progress"], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=errors, env=env) + process.stdin.write(json.dumps({"repo_path": repo, "mode": "fast"}).encode()) + process.stdin.close() + process.stdin = None + deadline = time.monotonic() + 120 + # Extraction is after begin_work released startup-v2. Observe that + # event before acquiring the real one-byte native startup lock. + while time.monotonic() < deadline: + trace = read_trace() + if b"msg=parallel.extract" in trace or b"msg=pass.definitions" in trace: + gate.seek(0) + try: + msvcrt.locking(gate.fileno(), msvcrt.LK_NBLCK, 1) + except OSError: + time.sleep(0.01) + continue + locked = True + break + if process.poll() is not None: + raise SetupFailure("index exited before the gate collision:\n" + + excerpt(trace)) + time.sleep(0.01) + if not locked: + raise SetupFailure("index never entered local work:\n" + excerpt(trace)) + while time.monotonic() < deadline: + trace = read_trace() + if b"Completed index_repository (" in trace: + break + if process.poll() is not None: + print("RED: index exited while a live peer held startup-v2:\n" + + excerpt(trace)) + return False + time.sleep(0.01) + else: + raise SetupFailure("CLI did not finish its work:\n" + excerpt(trace)) + # The outer CLI flushes this completion message after receiving + # the worker result and closing its daemon session, immediately + # before local cleanup. pipeline.done is too early: worker result + # handling can outlast a short lock hold. Keep the gate for longer + # than the old 500 ms cleanup deadline after this real boundary. + time.sleep(1) + stayed_alive = process.poll() is None + gate.seek(0) + msvcrt.locking(gate.fileno(), msvcrt.LK_UNLCK, 1) + locked = False + stdout, _ = process.communicate(timeout=30) + trace = read_trace() + if not stayed_alive or process.returncode != 0 or b"cleanup_timeout" in trace: + print("RED: completed work must wait out startup contention and exit 0 " + "after release (rc=%s):\n%s" % + (process.returncode, excerpt(stdout + trace))) + return False + print("PASS: completed local work waited for the live startup owner and exited 0") + return True + finally: + if locked: + gate.seek(0) + msvcrt.locking(gate.fileno(), msvcrt.LK_UNLCK, 1) + if process is not None and process.poll() is None: + subprocess.run(["taskkill", "/F", "/T", "/PID", str(process.pid)], + capture_output=True, timeout=30) + process.wait(timeout=30) + status = run_cli(binary, cache, ["daemon", "status"], timeout=30) + kill_pid(pid_from(out_text(status))) + + def section_cold_storm(binary, work): cache = section_dirs(work, "storm") results = _parallel_one_shots(binary, cache, 6) @@ -458,6 +577,7 @@ def main(): section_stop_refuses_busy, section_crash_recovery, section_churn_stability, + section_local_cleanup_waits_for_startup_gate, section_cold_storm, ] failed = []