Skip to content
100 changes: 98 additions & 2 deletions src/daemon/host.c
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,9 @@ enum {
HOST_HTTP_RETRY_INITIAL_MS = 1000,
HOST_HTTP_RETRY_MAX_MS = 30000,
HOST_WATCH_INTERVAL_MS = 5000,
/* Age-based temp cleaners need days of inactivity (macOS tmp_cleaner: 3);
* a minute bounds the blind window after a long sleep (#2178). */
HOST_COORDINATION_TOUCH_MS = 60000,
HOST_CONFLICT_LOG_CAP = 1024 * 1024,
HOST_OPERATION_LOG_CAP = 5 * 1024 * 1024,
HOST_PATH_CAP = 4096,
Expand Down Expand Up @@ -868,10 +871,58 @@ static bool host_background_start(host_state_t *host) {
return true;
}

/* Coordination handles the host retains for the generation's whole lifetime. */
typedef struct {
const cbm_daemon_ipc_endpoint_t *endpoint;
cbm_version_cohort_lease_t *cohort_lease;
cbm_version_cohort_daemon_claim_t *daemon_claim;
cbm_daemon_ipc_participant_guard_t *participant_guard;
} host_coordination_t;

typedef struct {
const char *lost; /* first held file whose path no longer names it */
const char *transient; /* first still-valid file that could not be refreshed */
} host_touch_result_t;

static void host_touch_note(host_touch_result_t *result, const char *file, int status) {
if (status == 0 && !result->lost) {
result->lost = file;
} else if (status < 0 && !result->transient) {
result->transient = file;
}
}

static int host_cohort_touch_status(cbm_version_cohort_status_t status) {
return status == CBM_VERSION_COHORT_OK ? 1 : (status == CBM_VERSION_COHORT_IO ? -1 : 0);
}

/* Refresh every long-held runtime file. A held lock whose path now names
* another inode coordinates nothing: forked index workers read the daemon as
* uncoordinated and a peer can start a second generation (#2178). Every file
* is visited even after one fails so the rest keep being refreshed. */
static host_touch_result_t host_coordination_touch(const host_coordination_t *coordination,
cbm_daemon_runtime_service_t *service) {
host_touch_result_t result = {0};
host_touch_note(&result, "cohort_daemon_claim",
host_cohort_touch_status(
cbm_version_cohort_daemon_claim_touch(coordination->daemon_claim)));
host_touch_note(
&result, "cohort_lease",
host_cohort_touch_status(cbm_version_cohort_lease_touch(coordination->cohort_lease)));
host_touch_note(&result, "participant_guard",
cbm_daemon_ipc_participant_guard_touch(coordination->endpoint,
coordination->participant_guard));
host_touch_note(&result, "listener", cbm_daemon_runtime_service_touch_listener(service));
return result;
}

static bool host_wait_for_lifetime(cbm_daemon_runtime_service_t *service,
atomic_int *stop_requested, host_state_t *host, bool permanent) {
atomic_int *stop_requested, host_state_t *host, bool permanent,
const host_coordination_t *coordination) {
uint64_t initial_deadline = cbm_now_ms() + HOST_INITIAL_CLIENT_TIMEOUT_MS;
uint64_t stopping_deadline = 0;
uint64_t next_touch = 0;
bool touch_degraded = false;
for (;;) {
cbm_daemon_runtime_service_state_t state = cbm_daemon_runtime_service_state(service);
if (state == CBM_DAEMON_RUNTIME_SERVICE_EXITED) {
Expand Down Expand Up @@ -903,6 +954,28 @@ static bool host_wait_for_lifetime(cbm_daemon_runtime_service_t *service,
stop ? "stop_requested" : "initial_window_expired");
return cbm_daemon_runtime_service_stop(service, HOST_RUNTIME_SHUTDOWN_MS);
}
/* Permanent generations too: exiting lets the next client start a
* coordinated generation, while re-claiming would race that client. */
if (cbm_now_ms() >= next_touch) {
next_touch = cbm_now_ms() + HOST_COORDINATION_TOUCH_MS;
host_touch_result_t touched = host_coordination_touch(coordination, service);
if (touched.lost) {
cbm_log_warn("daemon.lifetime_end", "reason", "coordination_file_lost", "file",
touched.lost);
return cbm_daemon_runtime_service_stop(service, HOST_RUNTIME_SHUTDOWN_MS);
}
/* A refresh failure on a still-valid file (read-only remount,
* EMFILE) is not loss: keep serving, retry next tick, and log
* only transitions so a persistent condition cannot flood. */
if ((touched.transient != NULL) != touch_degraded) {
touch_degraded = touched.transient != NULL;
if (touch_degraded) {
cbm_log_warn("daemon.coordination_touch_failed", "file", touched.transient);
} else {
cbm_log_info("daemon.coordination_touch_recovered");
}
}
}
host_http_reconcile_at(host, cbm_now_ms(), false);
/* Retire an ephemeral generation that lingered for cold-storm cohort
* participants once they drain, or once its bounded linger elapses.
Expand All @@ -912,6 +985,22 @@ static bool host_wait_for_lifetime(cbm_daemon_runtime_service_t *service,
}
}

bool cbm_daemon_host_wait_for_lifetime_for_test(cbm_daemon_runtime_service_t *service,
const cbm_daemon_ipc_endpoint_t *endpoint,
cbm_version_cohort_lease_t *cohort_lease,
cbm_version_cohort_daemon_claim_t *daemon_claim,
cbm_daemon_ipc_participant_guard_t *guard) {
host_state_t host = {0};
atomic_int stop_requested = 0;
host_coordination_t coordination = {
.endpoint = endpoint,
.cohort_lease = cohort_lease,
.daemon_claim = daemon_claim,
.participant_guard = guard,
};
return host_wait_for_lifetime(service, &stop_requested, &host, false, &coordination);
}

static bool host_application_shutdown(host_state_t *host) {
if (cbm_daemon_application_shutdown(host->application, HOST_APPLICATION_SHUTDOWN_MS)) {
return true;
Expand Down Expand Up @@ -1123,7 +1212,14 @@ int cbm_daemon_host_run(const cbm_daemon_host_config_t *config) {
: "unavailable",
"memory_budget_bytes", memory_budget, "physical_job_limit", physical_job_limit,
"worker_memory_budget_bytes", worker_memory_budget);
if (!host_wait_for_lifetime(service, config->stop_requested, &host, config->permanent)) {
host_coordination_t coordination = {
.endpoint = config->endpoint,
.cohort_lease = cohort_lease,
.daemon_claim = daemon_claim,
.participant_guard = participant_guard,
};
if (!host_wait_for_lifetime(service, config->stop_requested, &host, config->permanent,
&coordination)) {
host_force_terminate("runtime");
}

Expand Down
14 changes: 14 additions & 0 deletions src/daemon/host_internal.h
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,10 @@
#include <stdint.h>

struct cbm_daemon_ipc_endpoint;
struct cbm_daemon_ipc_participant_guard;
struct cbm_daemon_runtime_service;
struct cbm_version_cohort_daemon_claim;
struct cbm_version_cohort_lease;

typedef bool (*cbm_daemon_host_cleanup_release_for_test_fn)(void *context);

Expand Down Expand Up @@ -68,4 +72,14 @@ bool cbm_daemon_host_http_reconcile_free_refusal_for_test(
* the failure, the adapter must cancel so the final free succeeds. */
bool cbm_daemon_host_http_thread_create_failure_lifecycle_for_test(void);

/* Run the production lifetime loop over caller-held coordination handles with
* no HTTP host state and no stop request. The generation is ephemeral, so a
* loop that misses a loss still ends at the initial client window instead of
* hanging the test. Returns the loop's own result. */
bool cbm_daemon_host_wait_for_lifetime_for_test(
struct cbm_daemon_runtime_service *service, const struct cbm_daemon_ipc_endpoint *endpoint,
struct cbm_version_cohort_lease *cohort_lease,
struct cbm_version_cohort_daemon_claim *daemon_claim,
struct cbm_daemon_ipc_participant_guard *guard);

#endif /* CBM_DAEMON_HOST_INTERNAL_H */
109 changes: 109 additions & 0 deletions src/daemon/ipc.c
Original file line number Diff line number Diff line change
Expand Up @@ -3220,6 +3220,94 @@ void cbm_daemon_ipc_listener_close(cbm_daemon_ipc_listener_t *listener) {
free(listener);
}

/* Touch results: 1 refreshed, 0 lost (the path no longer names the owned
* file), -1 transient (a still-valid file could not be refreshed). Touch
* through the held descriptor only after proving the path still names it:
* refreshing by path would keep a replacement inode fresh and hide the loss
* (#2178). */
static int posix_touch_fold(int result, int next) {
return result == 0 || next == 0 ? 0 : (result < 0 || next < 0 ? -1 : 1);
}

/* Whether base_name still names the regular file (device, inode): 1 yes,
* 0 proven not (absent, another inode, a symlink or other non-regular
* entry), -1 undetermined (any other lookup failure, e.g. EIO or ESTALE). */
static int posix_path_names_inode(int directory_fd, const char *base_name, dev_t device,
ino_t inode) {
struct stat by_path;
if (fstatat(directory_fd, base_name, &by_path, AT_SYMLINK_NOFOLLOW) != 0) {
return errno == ENOENT ? 0 : -1;
}
return S_ISREG(by_path.st_mode) && by_path.st_dev == device && by_path.st_ino == inode ? 1 : 0;
}

static int posix_held_file_touch(int directory_fd, const char *base_name, int fd) {
struct stat by_handle;
if (fstat(fd, &by_handle) != 0) {
return -1;
}
int identity =
by_handle.st_nlink == 0
? 0
: posix_path_names_inode(directory_fd, base_name, by_handle.st_dev, by_handle.st_ino);
if (identity <= 0) {
return identity;
}
return futimens(fd, NULL) == 0 ? 1 : -1;
}

static int posix_held_lock_touch(int directory_fd, const process_lock_entry_t *entry, int fd) {
return entry ? posix_held_file_touch(directory_fd, entry->lock_name, fd) : 0;
}

static int posix_identity_marker_touch(const cbm_daemon_ipc_listener_t *listener) {
/* The marker is not held open. Classify by path first, so a symlink or
* other replacement is loss; only then reopen it to refresh, where a
* failure (EMFILE, EACCES) says nothing about the file and is transient. */
int identity = posix_path_names_inode(listener->dir_fd, listener->socket_identity_name,
listener->identity_device, listener->identity_inode);
if (identity <= 0) {
return identity;
}
int marker_fd = openat(listener->dir_fd, listener->socket_identity_name,
O_RDONLY | O_CLOEXEC | O_NOFOLLOW | O_NONBLOCK);
if (marker_fd < 0) {
return errno == ENOENT ? 0 : -1;
}
struct stat marker_status;
int result = -1;
if (fstat(marker_fd, &marker_status) == 0) {
/* Replaced between the lookup and the open. */
result = marker_status.st_dev != listener->identity_device ||
marker_status.st_ino != listener->identity_inode
? 0
: (futimens(marker_fd, NULL) == 0 ? 1 : -1);
}
(void)close(marker_fd);
return result;
}

int cbm_daemon_ipc_listener_touch(cbm_daemon_ipc_listener_t *listener,
cbm_daemon_ipc_participant_guard_t *external_guard) {
if (!listener || listener->dir_fd < 0 || listener->owner_pid != getpid() ||
!listener->lifetime_reservation) {
return 0;
}
int result =
posix_held_lock_touch(listener->dir_fd, listener->lifetime_reservation->process_entry,
listener->lifetime_reservation->fd);
const cbm_daemon_ipc_participant_guard_t *guards[] = {listener->participant_guard,
external_guard};
for (size_t i = 0; i < sizeof(guards) / sizeof(guards[0]); i++) {
if (guards[i]) {
result = posix_touch_fold(result, posix_held_lock_touch(listener->dir_fd,
guards[i]->legacy_process_entry,
guards[i]->legacy_fd));
}
}
return posix_touch_fold(result, posix_identity_marker_touch(listener));
}

int cbm_daemon_ipc_accept(cbm_daemon_ipc_listener_t *listener, uint32_t timeout_ms,
cbm_daemon_ipc_connection_t **connection_out) {
if (connection_out) {
Expand Down Expand Up @@ -3603,6 +3691,14 @@ bool cbm_daemon_ipc_participant_guard_release(cbm_daemon_ipc_participant_guard_t
return true;
}

int cbm_daemon_ipc_participant_guard_touch(const cbm_daemon_ipc_endpoint_t *endpoint,
cbm_daemon_ipc_participant_guard_t *guard) {
if (!endpoint || !guard || guard->owner_pid != getpid()) {
return 0;
}
return posix_held_lock_touch(endpoint->dir_fd, guard->legacy_process_entry, guard->legacy_fd);
}

int cbm_daemon_ipc_local_transition_try_acquire(
const cbm_daemon_ipc_endpoint_t *endpoint, cbm_daemon_ipc_local_transition_t **transition_out) {
if (transition_out) {
Expand Down Expand Up @@ -6746,6 +6842,19 @@ bool cbm_daemon_ipc_participant_guard_release(cbm_daemon_ipc_participant_guard_t
return true;
}

/* Windows has no age-based cleaner of the private runtime directory, and its
* held handles deny deletion; the #2178 heartbeat has nothing to refresh. */
int cbm_daemon_ipc_listener_touch(cbm_daemon_ipc_listener_t *listener,
cbm_daemon_ipc_participant_guard_t *external_guard) {
(void)external_guard;
return listener ? 1 : 0;
}

int cbm_daemon_ipc_participant_guard_touch(const cbm_daemon_ipc_endpoint_t *endpoint,
cbm_daemon_ipc_participant_guard_t *guard) {
return endpoint && guard ? 1 : 0;
}

int cbm_daemon_ipc_local_transition_try_acquire(
const cbm_daemon_ipc_endpoint_t *endpoint, cbm_daemon_ipc_local_transition_t **transition_out) {
if (transition_out) {
Expand Down
17 changes: 17 additions & 0 deletions src/daemon/ipc.h
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,19 @@ cbm_daemon_ipc_listener_t *cbm_daemon_ipc_listen_reserved(
cbm_daemon_ipc_lifetime_reservation_t **reservation_io);
void cbm_daemon_ipc_listener_close(cbm_daemon_ipc_listener_t *listener);

/* Heartbeat for the daemon's long-lived runtime artifacts: refresh the held
* lifetime reservation, the listener's participant guard (if it owns one),
* external_guard (optional, a guard the caller retains beside this listener)
* and the published identity marker so an age-based temp cleaner never
* deletes them (#2178). Returns 1 when all were refreshed, 0 once any no
* longer names the file this listener owns (a lost lifetime file lets a peer
* start a second generation; a lost identity marker makes listener close
* leave the socket pair behind), and -1 when a still-valid file could not be
* refreshed (transient). Only the thread that closes the listener may call it
* while the listener is open. */
int cbm_daemon_ipc_listener_touch(cbm_daemon_ipc_listener_t *listener,
cbm_daemon_ipc_participant_guard_t *external_guard);

/* Create or validate one private current-user directory at an already
* canonical local path. Ancestors are handle-validated without mutation:
* POSIX permits only root/current-user owners and no group/other write (apart
Expand Down Expand Up @@ -270,6 +283,10 @@ bool cbm_daemon_ipc_startup_lock_release(cbm_daemon_ipc_startup_lock_t **lock_io
int cbm_daemon_ipc_participant_guard_try_join(const cbm_daemon_ipc_endpoint_t *endpoint,
cbm_daemon_ipc_participant_guard_t **guard_out);
bool cbm_daemon_ipc_participant_guard_release(cbm_daemon_ipc_participant_guard_t **guard_io);
/* Same heartbeat contract and 1/0/-1 result as cbm_daemon_ipc_listener_touch,
* for a guard the caller retains itself. */
int cbm_daemon_ipc_participant_guard_touch(const cbm_daemon_ipc_endpoint_t *endpoint,
cbm_daemon_ipc_participant_guard_t *guard);

/* Standalone CLI work joins the legacy-compatible current participant group
* without becoming a daemon client. Acquisition retains startup-v2 only for
Expand Down
6 changes: 6 additions & 0 deletions src/daemon/runtime.c
Original file line number Diff line number Diff line change
Expand Up @@ -2578,6 +2578,12 @@ size_t cbm_daemon_runtime_service_active_connections(cbm_daemon_runtime_service_
return count;
}

int cbm_daemon_runtime_service_touch_listener(cbm_daemon_runtime_service_t *service) {
return service && service->listener
? cbm_daemon_ipc_listener_touch(service->listener, service->owned_participant_guard)
: 0;
}

void cbm_daemon_runtime_service_reconcile_lifetime(cbm_daemon_runtime_service_t *service) {
if (!service) {
return;
Expand Down
6 changes: 6 additions & 0 deletions src/daemon/runtime.h
Original file line number Diff line number Diff line change
Expand Up @@ -336,6 +336,12 @@ size_t cbm_daemon_runtime_service_active_connections(cbm_daemon_runtime_service_
* an unbounded idle hang. The host lifetime loop calls this every tick; it is a
* no-op unless a last-committed-client linger is armed. */
void cbm_daemon_runtime_service_reconcile_lifetime(cbm_daemon_runtime_service_t *service);
/* #2178 heartbeat over the listener's runtime artifacts, plus the participant
* guard the service owns when it started itself; see
* cbm_daemon_ipc_listener_touch for the 1/0/-1 result. Call only from the
* thread that stops and frees the service, which is the only thread that
* closes the listener. */
int cbm_daemon_runtime_service_touch_listener(cbm_daemon_runtime_service_t *service);
size_t cbm_daemon_runtime_service_job_subscribers(cbm_daemon_runtime_service_t *service,
const char *project_key);
uint64_t cbm_daemon_runtime_service_client_process_id(cbm_daemon_runtime_service_t *service,
Expand Down
31 changes: 31 additions & 0 deletions src/daemon/version_cohort.c
Original file line number Diff line number Diff line change
Expand Up @@ -363,6 +363,37 @@ cbm_private_file_lock_status_t cbm_version_cohort_lease_release(
return result;
}

/* Folds one lock's touch into an accumulated result: a lost file (UNSAFE)
* outranks a transient refresh failure (IO), which outranks OK. A lease holds
* only lifetime after admission, so absent locks are skipped. */
static cbm_version_cohort_status_t version_cohort_lock_touch(cbm_private_file_lock_t *lock,
cbm_version_cohort_status_t result) {
if (!lock || result == CBM_VERSION_COHORT_UNSAFE) {
return result;
}
cbm_private_file_lock_status_t status = cbm_private_file_lock_touch(lock);
if (status == CBM_PRIVATE_FILE_LOCK_UNSAFE) {
return CBM_VERSION_COHORT_UNSAFE;
}
return status == CBM_PRIVATE_FILE_LOCK_OK ? result : CBM_VERSION_COHORT_IO;
}

cbm_version_cohort_status_t cbm_version_cohort_lease_touch(cbm_version_cohort_lease_t *lease) {
if (!lease || !lease->lifetime) {
return CBM_VERSION_COHORT_UNSAFE;
}
cbm_version_cohort_status_t result =
version_cohort_lock_touch(lease->lifetime, CBM_VERSION_COHORT_OK);
result = version_cohort_lock_touch(lease->admission, result);
return version_cohort_lock_touch(lease->maintenance, result);
}

cbm_version_cohort_status_t cbm_version_cohort_daemon_claim_touch(
cbm_version_cohort_daemon_claim_t *claim) {
return claim && claim->marker ? version_cohort_lock_touch(claim->marker, CBM_VERSION_COHORT_OK)
: CBM_VERSION_COHORT_UNSAFE;
}

static cbm_version_cohort_status_t version_cohort_failed(cbm_version_cohort_lease_t *lease,
cbm_version_cohort_status_t status,
cbm_version_cohort_lease_t **lease_out) {
Expand Down
Loading
Loading