Merge pull request #2599 from pcristin/fix/windows-local-transition-cleanup

fix(daemon): wait out Windows local teardown contention

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) {

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 */

src/main.c

@@ -1645,20 +1645,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;

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);

tests/windows/test_daemon_stability.py

@@ -520,6 +520,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)
@@ -558,6 +677,7 @@ def main():
         section_stop_refuses_busy,
         section_crash_recovery,
         section_churn_stability,
+        section_local_cleanup_waits_for_startup_gate,
         section_cold_storm,
     ]
     failed = []