Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
81 changes: 76 additions & 5 deletions src/daemon/version_cohort.c
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down Expand Up @@ -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;
}
Expand Down Expand Up @@ -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,
Expand Down
19 changes: 15 additions & 4 deletions src/daemon/version_cohort.h
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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 */
133 changes: 133 additions & 0 deletions tests/test_version_cohort.c
Original file line number Diff line number Diff line change
Expand Up @@ -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"));
Expand Down Expand Up @@ -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);
Expand Down
Loading