Merge pull request #2047 from DeusData/fix/2046-cohort-conflict-retry-until-deadline

fix(daemon): retry a mismatched cohort holder until the caller's deadline

src/daemon/version_cohort.c

@@ -7,6 +7,7 @@
 #include "foundation/platform.h"
 #include "foundation/private_file_lock_internal.h"
 
+#include <stdatomic.h>
 #include <stdbool.h>
 #include <stdint.h>
 #include <stdio.h>
@@ -414,11 +415,10 @@ static cbm_version_cohort_status_t version_cohort_claim_new(
                                   CBM_PRIVATE_FILE_LOCK_SH, deadline_ms, &lease->lifetime));
 }
 
-cbm_version_cohort_status_t cbm_version_cohort_acquire(cbm_version_cohort_manager_t *manager,
-                                                       const cbm_daemon_build_identity_t *identity,
-                                                       uint64_t deadline_ms,
-                                                       cbm_version_cohort_lease_t **lease_out,
-                                                       cbm_daemon_conflict_t *conflict_out) {
+static cbm_version_cohort_status_t version_cohort_acquire_once(
+    cbm_version_cohort_manager_t *manager, const cbm_daemon_build_identity_t *identity,
+    uint64_t deadline_ms, cbm_version_cohort_lease_t **lease_out,
+    cbm_daemon_conflict_t *conflict_out) {
     if (lease_out) {
         *lease_out = NULL;
     }
@@ -569,6 +569,77 @@ cbm_version_cohort_status_t cbm_version_cohort_acquire(cbm_version_cohort_manage
     return CBM_VERSION_COHORT_OK;
 }
 
+#ifdef CBM_ENABLE_TEST_SEAMS
+/* #2046 test seam: counts conflict retries so a test can prove the waiting
+ * participant actually met the mismatched holder before that holder left —
+ * without it, "admitted after release" is indistinguishable from "arrived
+ * after release" and the test passes vacuously. */
+static atomic_uint_fast64_t version_cohort_conflict_retry_seam;
+uint64_t cbm_version_cohort_conflict_retries_for_testing(void) {
+    return atomic_load_explicit(&version_cohort_conflict_retry_seam, memory_order_acquire);
+}
+static void version_cohort_note_conflict_retry(void) {
+    (void)atomic_fetch_add_explicit(&version_cohort_conflict_retry_seam, 1, memory_order_release);
+}
+#else
+static void version_cohort_note_conflict_retry(void) {}
+#endif
+
+static const char *version_cohort_conflict_kind(const cbm_daemon_conflict_t *conflict) {
+    switch (conflict->status) {
+    case CBM_DAEMON_HELLO_VERSION_CONFLICT:
+        return "version";
+    case CBM_DAEMON_HELLO_BUILD_CONFLICT:
+        return "build";
+    case CBM_DAEMON_HELLO_PROTOCOL_ABI_CONFLICT:
+        return "protocol_abi";
+    case CBM_DAEMON_HELLO_STORE_ABI_CONFLICT:
+        return "store_abi";
+    case CBM_DAEMON_HELLO_FEATURE_ABI_CONFLICT:
+        return "feature_abi";
+    case CBM_DAEMON_HELLO_CACHE_CONFLICT:
+        return "cache_root";
+    default:
+        return "unknown";
+    }
+}
+
+/* A mismatched live holder is not necessarily a peer that will stay: an
+ * internal daemon keeps its lifetime SH lock for the few hundred milliseconds
+ * between its last client leaving and its teardown finishing, and a CLI under
+ * another CBM_CACHE_DIR arriving inside that window used to be refused with a
+ * cache-root conflict although nothing was running any more (#2046; the
+ * refusal was masked only by the local CLI's own slow startup). host.c already
+ * waits out the same handoff for the daemon claim marker. Admission therefore
+ * retries a conflict until the caller's deadline, holding NO guard between
+ * attempts so compatible peers and activations are never queued behind the
+ * wait. On deadline the last observed conflict is reported exactly as before;
+ * an indefinite deadline keeps failing fast because a mismatched holder may
+ * never leave. */
+cbm_version_cohort_status_t cbm_version_cohort_acquire(cbm_version_cohort_manager_t *manager,
+                                                       const cbm_daemon_build_identity_t *identity,
+                                                       uint64_t deadline_ms,
+                                                       cbm_version_cohort_lease_t **lease_out,
+                                                       cbm_daemon_conflict_t *conflict_out) {
+    bool retry_logged = false;
+    for (;;) {
+        cbm_version_cohort_status_t status =
+            version_cohort_acquire_once(manager, identity, deadline_ms, lease_out, conflict_out);
+        if (status != CBM_VERSION_COHORT_CONFLICT || deadline_ms == UINT64_MAX ||
+            cbm_now_ms() >= deadline_ms) {
+            return status;
+        }
+        if (!retry_logged) {
+            retry_logged = true;
+            cbm_log_info("version_cohort.conflict_retry", "reason",
+                         version_cohort_conflict_kind(conflict_out), "active_build",
+                         conflict_out->active_build_fingerprint);
+        }
+        version_cohort_note_conflict_retry();
+        version_cohort_retry_sleep();
+    }
+}
+
 static cbm_version_cohort_status_t version_cohort_reserve_for_mutation_internal(
     cbm_version_cohort_manager_t *manager, uint64_t deadline_ms,
     cbm_version_cohort_quiesce_fn quiesce, void *quiesce_context,

src/daemon/version_cohort.h

@@ -58,10 +58,14 @@ cbm_version_cohort_manager_t *cbm_version_cohort_manager_new(
 /* Admission first takes the maintenance gate SH without waiting, holds it
  * across the short admission EX transition, then retains SH on the cohort
  * lifetime file. Active maintenance therefore fails fast with BUSY. Exact
- * identity peers share SH; a different version, build, or ABI returns
- * CONFLICT with conflict_out populated. deadline_ms is an absolute
- * cbm_now_ms() deadline; UINT64_MAX waits indefinitely. Every non-NULL
- * lease_out, including cleanup-only IO state, must be released. */
+ * identity peers share SH; a different version, build, ABI, or cache root
+ * is a CONFLICT — retried, with every guard released between attempts, until
+ * deadline_ms, because the mismatched holder is often a draining daemon or a
+ * short-lived CLI that is about to leave (#2046). Only when the deadline
+ * passes is CONFLICT returned with conflict_out populated. deadline_ms is an
+ * absolute cbm_now_ms() deadline; UINT64_MAX waits indefinitely for locks but
+ * never for a conflicting holder. Every non-NULL lease_out, including
+ * cleanup-only IO state, must be released. */
 cbm_version_cohort_status_t cbm_version_cohort_acquire(cbm_version_cohort_manager_t *manager,
                                                        const cbm_daemon_build_identity_t *identity,
                                                        uint64_t deadline_ms,
@@ -153,4 +157,11 @@ bool cbm_version_cohort_log_conflict(const cbm_daemon_conflict_t *conflict);
  * is active but no current-generation coordination marker can be verified. */
 bool cbm_version_cohort_log_uncoordinated_daemon(const cbm_daemon_build_identity_t *requested);
 
+#ifdef CBM_ENABLE_TEST_SEAMS
+/* Monotonic count of admission attempts that met a mismatched live holder and
+ * retried (#2046). Lets a test prove the waiter observed the conflict before
+ * the holder left, instead of passing vacuously by arriving late. */
+uint64_t cbm_version_cohort_conflict_retries_for_testing(void);
+#endif
+
 #endif /* CBM_DAEMON_VERSION_COHORT_H */

tests/test_version_cohort.c

@@ -265,6 +265,137 @@ TEST(version_cohort_rejects_exact_build_with_different_cache_root) {
     PASS();
 }
 
+typedef struct {
+    cbm_version_cohort_manager_t *manager;
+    cbm_daemon_build_identity_t identity;
+    uint64_t deadline_ms;
+    cbm_version_cohort_status_t status;
+    cbm_version_cohort_lease_t *lease;
+    cbm_daemon_conflict_t conflict;
+    atomic_bool finished;
+} version_cohort_acquire_wait_t;
+
+static void *version_cohort_acquire_wait_thread(void *context) {
+    version_cohort_acquire_wait_t *wait = context;
+    wait->status = cbm_version_cohort_acquire(wait->manager, &wait->identity, wait->deadline_ms,
+                                              &wait->lease, &wait->conflict);
+    atomic_store_explicit(&wait->finished, true, memory_order_release);
+    return NULL;
+}
+
+static bool version_cohort_wait_for_retries_above(uint64_t baseline, uint64_t deadline_ms) {
+    while (cbm_version_cohort_conflict_retries_for_testing() <= baseline &&
+           cbm_now_ms() < deadline_ms) {
+        cbm_usleep(1000);
+    }
+    return cbm_version_cohort_conflict_retries_for_testing() > baseline;
+}
+
+/* #2046: a mismatched holder that never leaves is still a conflict — but the
+ * caller's deadline is the budget for waiting it out, so the refusal may only
+ * arrive once that budget is spent, and must carry the same detail as before.
+ * The lower bound is the contract under test, not a timing guess: the holder
+ * is released only after the call returns. */
+TEST(version_cohort_conflict_is_retried_until_the_deadline) {
+    version_cohort_fixture_t fixture;
+    ASSERT_TRUE(version_cohort_fixture_start(&fixture, "conflict-deadline"));
+    cbm_version_cohort_manager_t *first = cbm_version_cohort_manager_new(fixture.endpoint);
+    cbm_version_cohort_manager_t *second = cbm_version_cohort_manager_new(fixture.endpoint);
+    ASSERT_NOT_NULL(first);
+    ASSERT_NOT_NULL(second);
+
+    cbm_daemon_build_identity_t active = version_cohort_identity("2.4.0", VERSION_COHORT_BUILD_A);
+    cbm_daemon_build_identity_t requested = active;
+    active.cache_fingerprint = VERSION_COHORT_CACHE_A;
+    requested.cache_fingerprint = VERSION_COHORT_CACHE_B;
+    cbm_version_cohort_lease_t *active_lease = NULL;
+    cbm_version_cohort_lease_t *requested_lease = NULL;
+    cbm_daemon_conflict_t conflict;
+    ASSERT_EQ(cbm_version_cohort_acquire(first, &active, UINT64_MAX, &active_lease, &conflict),
+              CBM_VERSION_COHORT_OK);
+
+    uint64_t retries_before = cbm_version_cohort_conflict_retries_for_testing();
+    uint64_t deadline = cbm_now_ms() + 300U;
+    cbm_version_cohort_status_t status =
+        cbm_version_cohort_acquire(second, &requested, deadline, &requested_lease, &conflict);
+    uint64_t returned_at = cbm_now_ms();
+
+    ASSERT_EQ(status, CBM_VERSION_COHORT_CONFLICT);
+    ASSERT_NULL(requested_lease);
+    ASSERT_TRUE(returned_at >= deadline);
+    ASSERT_TRUE(cbm_version_cohort_conflict_retries_for_testing() > retries_before);
+    ASSERT_EQ(conflict.status, CBM_DAEMON_HELLO_CACHE_CONFLICT);
+    ASSERT_STR_EQ(conflict.active_cache_fingerprint, VERSION_COHORT_CACHE_A);
+    ASSERT_STR_EQ(conflict.requested_cache_fingerprint, VERSION_COHORT_CACHE_B);
+
+    /* An indefinite deadline never waits on a conflicting holder. */
+    retries_before = cbm_version_cohort_conflict_retries_for_testing();
+    ASSERT_EQ(cbm_version_cohort_acquire(second, &requested, UINT64_MAX, &requested_lease,
+                                         &conflict),
+              CBM_VERSION_COHORT_CONFLICT);
+    ASSERT_NULL(requested_lease);
+    ASSERT_EQ(cbm_version_cohort_conflict_retries_for_testing(), retries_before);
+
+    version_cohort_release(&active_lease);
+    version_cohort_manager_close(&second);
+    version_cohort_manager_close(&first);
+    version_cohort_fixture_finish(&fixture);
+    PASS();
+}
+
+/* #2046: the draining-daemon handoff. The waiter is admitted once the
+ * mismatched holder leaves — and the test releases that holder only after the
+ * seam proves the waiter already met it, so "admitted" cannot mean "arrived
+ * late". The waiter's deadline is a backstop against a hang, never the
+ * decider: the asserted state is OK-with-lease after the release. */
+TEST(version_cohort_conflict_waiter_is_admitted_when_the_holder_leaves) {
+    version_cohort_fixture_t fixture;
+    ASSERT_TRUE(version_cohort_fixture_start(&fixture, "conflict-handoff"));
+    cbm_version_cohort_manager_t *first = cbm_version_cohort_manager_new(fixture.endpoint);
+    cbm_version_cohort_manager_t *second = cbm_version_cohort_manager_new(fixture.endpoint);
+    ASSERT_NOT_NULL(first);
+    ASSERT_NOT_NULL(second);
+
+    cbm_daemon_build_identity_t active = version_cohort_identity("2.4.0", VERSION_COHORT_BUILD_A);
+    active.cache_fingerprint = VERSION_COHORT_CACHE_A;
+    cbm_version_cohort_lease_t *active_lease = NULL;
+    cbm_daemon_conflict_t conflict;
+    ASSERT_EQ(cbm_version_cohort_acquire(first, &active, UINT64_MAX, &active_lease, &conflict),
+              CBM_VERSION_COHORT_OK);
+
+    version_cohort_acquire_wait_t wait;
+    memset(&wait, 0, sizeof(wait));
+    wait.manager = second;
+    wait.identity = active;
+    wait.identity.cache_fingerprint = VERSION_COHORT_CACHE_B;
+    wait.deadline_ms = cbm_now_ms() + 10000U;
+    wait.status = CBM_VERSION_COHORT_IO;
+    atomic_init(&wait.finished, false);
+
+    uint64_t retries_before = cbm_version_cohort_conflict_retries_for_testing();
+    cbm_thread_t thread;
+    bool started = cbm_thread_create(&thread, 0, version_cohort_acquire_wait_thread, &wait) == 0;
+    bool met_holder =
+        started && version_cohort_wait_for_retries_above(retries_before, cbm_now_ms() + 5000U);
+    bool still_waiting = started && !atomic_load_explicit(&wait.finished, memory_order_acquire);
+    version_cohort_release(&active_lease);
+    bool finished = started && version_cohort_wait_for_atomic(&wait.finished, cbm_now_ms() + 5000U);
+    bool joined = started && cbm_thread_join(&thread) == 0;
+
+    version_cohort_release(&wait.lease);
+    version_cohort_manager_close(&second);
+    version_cohort_manager_close(&first);
+    version_cohort_fixture_finish(&fixture);
+
+    ASSERT_TRUE(started);
+    ASSERT_TRUE(met_holder);
+    ASSERT_TRUE(still_waiting);
+    ASSERT_TRUE(finished);
+    ASSERT_TRUE(joined);
+    ASSERT_EQ(wait.status, CBM_VERSION_COHORT_OK);
+    PASS();
+}
+
 TEST(version_cohort_exclusive_activation_blocks_and_is_blocked_by_participants) {
     version_cohort_fixture_t fixture;
     ASSERT_TRUE(version_cohort_fixture_start(&fixture, "activation"));
@@ -896,6 +1027,8 @@ SUITE(version_cohort) {
     RUN_TEST(version_cohort_rejects_same_hash_with_different_abi);
     RUN_TEST(version_cohort_rejects_missing_cache_fingerprint);
     RUN_TEST(version_cohort_rejects_exact_build_with_different_cache_root);
+    RUN_TEST(version_cohort_conflict_is_retried_until_the_deadline);
+    RUN_TEST(version_cohort_conflict_waiter_is_admitted_when_the_holder_leaves);
     RUN_TEST(version_cohort_exclusive_activation_blocks_and_is_blocked_by_participants);
     RUN_TEST(version_cohort_mutation_intent_fails_new_admission_and_spans_lease);
     RUN_TEST(version_cohort_mutation_waits_for_every_lifetime_participant);