mirror of
https://github.com/DeusData/codebase-memory-mcp.git
synced 2026-11-04 10:48:20 +00:00
Martin Vogel
bf30a1aaf0
The recovery path replays the session context and UI settings on the
replacement daemon. It waited on the bootstrap connect budget, which only
bounds a probe (1 s in production), while main.c gives the first
set_context and set_ui_config the startup budget since eb294298e. A
replacement daemon whose UI save runs longer than a probe would end an
otherwise good recovery.
frontend_configure_client now waits on startup_timeout_ms. A test seam
records the budget the replay used, and the recovery fixture gives the
bootstrap distinct connect and startup budgets and exits 75 when the
replay waited on the wrong one. With the old budget exactly the three
recovery tests fail with exit 75; with the fix daemon_frontend is 17/17.
Signed-off-by: Martin Vogel <martin.vogel.tech@gmail.com>
2002 lines
90 KiB
C
2002 lines
90 KiB
C
/*
|
|
* test_daemon_frontend.c — Exact JSON-RPC cancellation and maintenance.
|
|
*/
|
|
#include "test_framework.h"
|
|
#include "test_helpers.h"
|
|
|
|
#include "daemon/application.h"
|
|
#include "daemon/frontend.h"
|
|
#include "daemon/ipc.h"
|
|
#include "daemon/service.h"
|
|
#include "daemon/version_cohort.h"
|
|
#include "foundation/compat.h"
|
|
#include "foundation/compat_thread.h"
|
|
#include "foundation/platform.h"
|
|
#include "foundation/sanitized.h"
|
|
|
|
#include <stdint.h>
|
|
#include <stdatomic.h>
|
|
#include <stdio.h>
|
|
#include <string.h>
|
|
|
|
#ifndef _WIN32
|
|
#include <errno.h>
|
|
#include <fcntl.h>
|
|
#include <signal.h>
|
|
#include <sys/wait.h>
|
|
#include <unistd.h>
|
|
#endif
|
|
|
|
enum { FRONTEND_TEST_PATH_CAP = 1024 };
|
|
|
|
typedef struct {
|
|
char parent[FRONTEND_TEST_PATH_CAP];
|
|
cbm_daemon_ipc_endpoint_t *endpoint;
|
|
cbm_version_cohort_manager_t *manager;
|
|
} frontend_maintenance_fixture_t;
|
|
|
|
#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS
|
|
typedef struct {
|
|
frontend_maintenance_fixture_t maintenance;
|
|
char conflict_log[FRONTEND_TEST_PATH_CAP];
|
|
char build_fingerprint[CBM_DAEMON_BUILD_FINGERPRINT_SIZE];
|
|
cbm_daemon_build_identity_t identity;
|
|
cbm_daemon_runtime_service_t *service;
|
|
cbm_daemon_runtime_client_t *client;
|
|
} frontend_idle_fixture_t;
|
|
|
|
typedef struct {
|
|
cbm_daemon_runtime_client_t *client;
|
|
cbm_version_cohort_manager_t *manager;
|
|
const cbm_daemon_frontend_session_config_t *session;
|
|
FILE *input;
|
|
FILE *output;
|
|
int result;
|
|
} frontend_idle_run_t;
|
|
#endif
|
|
|
|
static bool frontend_maintenance_fixture_start(frontend_maintenance_fixture_t *fixture,
|
|
const char *tag) {
|
|
memset(fixture, 0, sizeof(*fixture));
|
|
/* Endpoint parents need production-shaped ancestry (LocalAppData on
|
|
* Windows); temp roots are refused by the runtime ancestry validation. */
|
|
if (!th_secure_runtime_parent_new(fixture->parent, sizeof(fixture->parent), tag)) {
|
|
return false;
|
|
}
|
|
fixture->endpoint = cbm_daemon_ipc_endpoint_new("0123456789abcdef", fixture->parent);
|
|
fixture->manager = fixture->endpoint ? cbm_version_cohort_manager_new(fixture->endpoint) : NULL;
|
|
return fixture->endpoint && fixture->manager;
|
|
}
|
|
|
|
static void frontend_maintenance_fixture_finish(frontend_maintenance_fixture_t *fixture) {
|
|
while (fixture->manager &&
|
|
cbm_version_cohort_manager_free(&fixture->manager) != CBM_PRIVATE_FILE_LOCK_OK) {
|
|
cbm_usleep(1000);
|
|
}
|
|
cbm_daemon_ipc_endpoint_free(fixture->endpoint);
|
|
if (fixture->parent[0]) {
|
|
(void)th_rmtree(fixture->parent);
|
|
}
|
|
memset(fixture, 0, sizeof(*fixture));
|
|
}
|
|
|
|
#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS
|
|
static cbm_daemon_runtime_application_session_t *frontend_idle_session_open(
|
|
void *context, cbm_daemon_client_id_t client_id, uint64_t authenticated_process_id) {
|
|
return context && client_id != CBM_DAEMON_CLIENT_ID_INVALID && authenticated_process_id != 0
|
|
? (cbm_daemon_runtime_application_session_t *)context
|
|
: NULL;
|
|
}
|
|
|
|
static cbm_daemon_runtime_application_status_t frontend_idle_request(
|
|
void *context, cbm_daemon_runtime_application_session_t *session,
|
|
cbm_daemon_runtime_application_token_t request_token, const uint8_t *request,
|
|
uint32_t request_length, uint8_t **response_out, uint32_t *response_length_out) {
|
|
(void)request_token;
|
|
if (!context || session != (cbm_daemon_runtime_application_session_t *)context ||
|
|
!response_out || !response_length_out || (request_length > 0 && !request)) {
|
|
return CBM_DAEMON_RUNTIME_APPLICATION_HANDLER_ERROR;
|
|
}
|
|
*response_out = NULL;
|
|
*response_length_out = 0;
|
|
return CBM_DAEMON_RUNTIME_APPLICATION_OK;
|
|
}
|
|
|
|
static void frontend_idle_request_cancel(void *context,
|
|
cbm_daemon_runtime_application_session_t *session,
|
|
cbm_daemon_runtime_application_token_t request_token) {
|
|
(void)context;
|
|
(void)session;
|
|
(void)request_token;
|
|
}
|
|
|
|
static void frontend_idle_session_cancel(void *context,
|
|
cbm_daemon_runtime_application_session_t *session) {
|
|
(void)context;
|
|
(void)session;
|
|
}
|
|
|
|
static void frontend_idle_session_close(void *context,
|
|
cbm_daemon_runtime_application_session_t *session) {
|
|
(void)context;
|
|
(void)session;
|
|
}
|
|
|
|
static uint64_t frontend_test_process_id(void) {
|
|
#ifdef _WIN32
|
|
return (uint64_t)GetCurrentProcessId();
|
|
#else
|
|
return (uint64_t)getpid();
|
|
#endif
|
|
}
|
|
|
|
static bool frontend_idle_fixture_start(frontend_idle_fixture_t *fixture) {
|
|
static const char cache[] = "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd";
|
|
memset(fixture, 0, sizeof(*fixture));
|
|
if (!frontend_maintenance_fixture_start(&fixture->maintenance, "idle-observer")) {
|
|
return false;
|
|
}
|
|
int log_written = snprintf(fixture->conflict_log, sizeof(fixture->conflict_log),
|
|
"%s/conflicts.ndjson", fixture->maintenance.parent);
|
|
if (log_written <= 0 || log_written >= (int)sizeof(fixture->conflict_log) ||
|
|
!cbm_daemon_runtime_process_build_fingerprint(frontend_test_process_id(),
|
|
fixture->build_fingerprint)) {
|
|
return false;
|
|
}
|
|
fixture->identity = (cbm_daemon_build_identity_t){
|
|
.semantic_version = "2.4.0",
|
|
.build_fingerprint = fixture->build_fingerprint,
|
|
.cache_fingerprint = cache,
|
|
.protocol_abi = 3,
|
|
.store_abi = 11,
|
|
.feature_abi = 7,
|
|
};
|
|
cbm_daemon_runtime_application_callbacks_t callbacks = {
|
|
.context = fixture,
|
|
.session_open = frontend_idle_session_open,
|
|
.request = frontend_idle_request,
|
|
.request_cancel = frontend_idle_request_cancel,
|
|
.session_cancel = frontend_idle_session_cancel,
|
|
.session_close = frontend_idle_session_close,
|
|
};
|
|
cbm_daemon_runtime_service_config_t config = {
|
|
.endpoint = fixture->maintenance.endpoint,
|
|
.identity = fixture->identity,
|
|
.conflict_log_path = fixture->conflict_log,
|
|
.conflict_log_cap_bytes = 64U * 1024U,
|
|
.max_clients = 2,
|
|
.lease_timeout_ms = 5000,
|
|
.request_timeout_ms = 30000,
|
|
.shutdown_timeout_ms = 30000,
|
|
.application = callbacks,
|
|
};
|
|
fixture->service = cbm_daemon_runtime_service_start(&config);
|
|
cbm_daemon_runtime_connect_result_t connect_result = {0};
|
|
fixture->client =
|
|
fixture->service
|
|
? cbm_daemon_runtime_client_connect(fixture->maintenance.endpoint, &fixture->identity,
|
|
30000, &connect_result)
|
|
: NULL;
|
|
return fixture->client && connect_result.status == CBM_DAEMON_RUNTIME_CONNECT_ACCEPTED;
|
|
}
|
|
|
|
static bool frontend_idle_fixture_finish(frontend_idle_fixture_t *fixture) {
|
|
bool ok = true;
|
|
if (fixture->client) {
|
|
ok = cbm_daemon_runtime_client_close(fixture->client, 30000) && ok;
|
|
fixture->client = NULL;
|
|
}
|
|
if (fixture->service) {
|
|
if (cbm_daemon_runtime_service_state(fixture->service) !=
|
|
CBM_DAEMON_RUNTIME_SERVICE_EXITED) {
|
|
ok = cbm_daemon_runtime_service_stop(fixture->service, 30000) && ok;
|
|
}
|
|
ok = cbm_daemon_runtime_service_free(fixture->service) && ok;
|
|
fixture->service = NULL;
|
|
}
|
|
frontend_maintenance_fixture_finish(&fixture->maintenance);
|
|
return ok;
|
|
}
|
|
|
|
static FILE *frontend_test_fdopen_read(int fd) {
|
|
#ifdef _WIN32
|
|
return _fdopen(fd, "rb");
|
|
#else
|
|
return fdopen(fd, "rb");
|
|
#endif
|
|
}
|
|
|
|
static int frontend_test_close_fd(int fd) {
|
|
#ifdef _WIN32
|
|
return _close(fd);
|
|
#else
|
|
return close(fd);
|
|
#endif
|
|
}
|
|
|
|
static void *frontend_idle_run(void *opaque) {
|
|
frontend_idle_run_t *run = opaque;
|
|
run->result = cbm_daemon_frontend_mcp_run(run->client, run->manager, run->session, run->input,
|
|
run->output);
|
|
return NULL;
|
|
}
|
|
#endif
|
|
|
|
#ifndef _WIN32
|
|
static const char FRONTEND_TEST_BUILD[] =
|
|
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
|
|
static const char FRONTEND_TEST_CACHE[] =
|
|
"cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc";
|
|
|
|
enum {
|
|
/* Hang guards, not benchmarks: starved 3-4-core CI runners need several
|
|
* seconds for frontend/daemon startup paths that finish instantly on a
|
|
* dev machine (see the index-supervisor calibration). */
|
|
FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS = 30000,
|
|
FRONTEND_EOF_TEST_CATASTROPHIC_TIMEOUT_S = 90,
|
|
/* Recovery can spend most of the generic guard restarting an instrumented
|
|
* daemon before the clean-EOF case enters its deliberate 15s drain. */
|
|
FRONTEND_RECOVERY_TEST_CATASTROPHIC_TIMEOUT_S = 120,
|
|
FRONTEND_BACKPRESSURE_MESSAGE_BYTES = 2 * 1024 * 1024,
|
|
FRONTEND_BACKPRESSURE_FRONTEND_TIMEOUT_S = 90,
|
|
FRONTEND_BACKPRESSURE_DAEMON_TIMEOUT_S = 90,
|
|
FRONTEND_BACKPRESSURE_RUNTIME_TIMEOUT_MS = 90000,
|
|
FRONTEND_BACKPRESSURE_CLEANUP_TIMEOUT_MS = 30000,
|
|
/* The production queue is deliberately bounded below this count. Keep the
|
|
* regression black-box: it must remain valid if the exact capacity changes
|
|
* while still proving that overload cannot hide an already-pending EOF. */
|
|
FRONTEND_EOF_TEST_OVERFLOW_MESSAGES = 32,
|
|
/* Hold replacement context past the production frontend's 15 second
|
|
* no-progress EOF
|
|
drain so shutdown deterministically overlaps recovery. */
|
|
FRONTEND_RECOVERY_EOF_RELEASE_MS = 16000,
|
|
/* Startup budget of the recovery bootstrap. It differs from the connect
|
|
* budget (FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS) so the recovery tests can
|
|
* tell which one the session replay waited on, and it is the larger one,
|
|
* as in production (1 s connect probe, 30 s startup). */
|
|
FRONTEND_RECOVERY_TEST_STARTUP_TIMEOUT_MS = 45000,
|
|
};
|
|
|
|
typedef enum {
|
|
FRONTEND_RECOVERY_RETRY = 0,
|
|
FRONTEND_RECOVERY_CANCEL = 1,
|
|
FRONTEND_RECOVERY_CLEAN_EOF = 2,
|
|
} frontend_recovery_mode_t;
|
|
|
|
typedef struct {
|
|
atomic_int requests;
|
|
atomic_int session_cancels;
|
|
atomic_int sessions_opened;
|
|
atomic_int first_session_cancels;
|
|
atomic_int second_session_cancels;
|
|
atomic_bool block_first_request;
|
|
atomic_bool first_request_started;
|
|
/* Overflow mode: set by the writer AFTER stdin is fully pumped and closed,
|
|
* releasing the deliberately-held first request. The hold is what forces
|
|
* the input queue to actually fill (proving enqueue backpressures instead
|
|
* of failing the session); the release is a state transition, not a
|
|
* timeout, so the drain that follows is deterministic. */
|
|
atomic_bool release_first_request;
|
|
int request_observed_fd;
|
|
int session_cancel_fd;
|
|
} frontend_eof_application_context_t;
|
|
|
|
typedef struct {
|
|
frontend_eof_application_context_t *context;
|
|
atomic_bool cancelled;
|
|
int session_index;
|
|
} frontend_eof_application_session_t;
|
|
|
|
typedef struct {
|
|
int fd;
|
|
bool overflow;
|
|
frontend_recovery_mode_t recovery_mode;
|
|
frontend_eof_application_context_t *application;
|
|
atomic_bool finished;
|
|
atomic_bool succeeded;
|
|
} frontend_eof_writer_t;
|
|
|
|
typedef struct {
|
|
char conflict_log[FRONTEND_TEST_PATH_CAP];
|
|
char build_fingerprint[CBM_DAEMON_BUILD_FINGERPRINT_SIZE];
|
|
cbm_daemon_ipc_endpoint_t *endpoint;
|
|
cbm_daemon_build_identity_t identity;
|
|
cbm_version_cohort_manager_t *manager;
|
|
cbm_daemon_runtime_service_t *service;
|
|
cbm_daemon_runtime_client_t *client;
|
|
frontend_eof_application_context_t application;
|
|
} frontend_eof_fixture_t;
|
|
|
|
static cbm_daemon_build_identity_t frontend_test_identity(void) {
|
|
cbm_daemon_build_identity_t identity = {
|
|
.semantic_version = "2.4.0",
|
|
.build_fingerprint = FRONTEND_TEST_BUILD,
|
|
.cache_fingerprint = FRONTEND_TEST_CACHE,
|
|
.protocol_abi = 3,
|
|
.store_abi = 11,
|
|
.feature_abi = 7,
|
|
};
|
|
return identity;
|
|
}
|
|
|
|
static cbm_daemon_runtime_application_session_t *frontend_eof_application_session_open(
|
|
void *opaque, cbm_daemon_client_id_t client_id, uint64_t authenticated_process_id) {
|
|
frontend_eof_application_context_t *context = opaque;
|
|
frontend_eof_application_session_t *session = calloc(1, sizeof(*session));
|
|
if (!context || !session || client_id == CBM_DAEMON_CLIENT_ID_INVALID ||
|
|
authenticated_process_id == 0) {
|
|
free(session);
|
|
return NULL;
|
|
}
|
|
session->context = context;
|
|
session->session_index =
|
|
atomic_fetch_add_explicit(&context->sessions_opened, 1, memory_order_relaxed);
|
|
atomic_init(&session->cancelled, false);
|
|
return (cbm_daemon_runtime_application_session_t *)session;
|
|
}
|
|
|
|
static cbm_daemon_runtime_application_status_t frontend_eof_application_request(
|
|
void *opaque, cbm_daemon_runtime_application_session_t *opaque_session,
|
|
cbm_daemon_runtime_application_token_t request_token, const uint8_t *request,
|
|
uint32_t request_length, uint8_t **response_out, uint32_t *response_length_out) {
|
|
(void)request_token;
|
|
frontend_eof_application_context_t *context = opaque;
|
|
frontend_eof_application_session_t *session =
|
|
(frontend_eof_application_session_t *)opaque_session;
|
|
if (!context || !session || session->context != context || !response_out ||
|
|
!response_length_out || (request_length > 0 && !request)) {
|
|
return CBM_DAEMON_RUNTIME_APPLICATION_HANDLER_ERROR;
|
|
}
|
|
*response_out = NULL;
|
|
*response_length_out = 0;
|
|
int request_index = atomic_fetch_add_explicit(&context->requests, 1, memory_order_relaxed);
|
|
if (request_index == 0 &&
|
|
atomic_load_explicit(&context->block_first_request, memory_order_acquire)) {
|
|
atomic_store_explicit(&context->first_request_started, true, memory_order_release);
|
|
while (!atomic_load_explicit(&session->cancelled, memory_order_acquire) &&
|
|
!atomic_load_explicit(&context->release_first_request, memory_order_acquire)) {
|
|
cbm_usleep(1000);
|
|
}
|
|
if (atomic_load_explicit(&session->cancelled, memory_order_acquire)) {
|
|
return CBM_DAEMON_RUNTIME_APPLICATION_CANCELLED;
|
|
}
|
|
/* Released: fall through and answer normally like every later item. */
|
|
}
|
|
if (request_length == 0 || request[0] == CBM_DAEMON_APPLICATION_REQUEST_SET_CONTEXT) {
|
|
return CBM_DAEMON_RUNTIME_APPLICATION_OK;
|
|
}
|
|
const uint8_t *response_source = request;
|
|
uint32_t response_length = request_length;
|
|
if (request[0] == CBM_DAEMON_APPLICATION_REQUEST_MCP) {
|
|
response_source++;
|
|
response_length--;
|
|
}
|
|
uint8_t *response = malloc(response_length);
|
|
if (!response) {
|
|
return CBM_DAEMON_RUNTIME_APPLICATION_HANDLER_ERROR;
|
|
}
|
|
memcpy(response, response_source, response_length);
|
|
*response_out = response;
|
|
*response_length_out = response_length;
|
|
if (context->request_observed_fd >= 0) {
|
|
const char marker = 'Q';
|
|
(void)write(context->request_observed_fd, &marker, 1);
|
|
}
|
|
return CBM_DAEMON_RUNTIME_APPLICATION_OK;
|
|
}
|
|
|
|
static void frontend_eof_application_request_cancel(
|
|
void *opaque, cbm_daemon_runtime_application_session_t *opaque_session,
|
|
cbm_daemon_runtime_application_token_t request_token) {
|
|
(void)request_token;
|
|
frontend_eof_application_context_t *context = opaque;
|
|
frontend_eof_application_session_t *session =
|
|
(frontend_eof_application_session_t *)opaque_session;
|
|
if (context && session && session->context == context) {
|
|
atomic_store_explicit(&session->cancelled, true, memory_order_release);
|
|
}
|
|
}
|
|
|
|
static void frontend_eof_application_session_cancel(
|
|
void *opaque, cbm_daemon_runtime_application_session_t *opaque_session) {
|
|
frontend_eof_application_context_t *context = opaque;
|
|
frontend_eof_application_session_t *session =
|
|
(frontend_eof_application_session_t *)opaque_session;
|
|
if (context && session && session->context == context) {
|
|
atomic_store_explicit(&session->cancelled, true, memory_order_release);
|
|
if (session->session_index == 0) {
|
|
(void)atomic_fetch_add_explicit(&context->first_session_cancels, 1,
|
|
memory_order_relaxed);
|
|
} else if (session->session_index == 1) {
|
|
(void)atomic_fetch_add_explicit(&context->second_session_cancels, 1,
|
|
memory_order_relaxed);
|
|
}
|
|
/* Publish the total only after the exact-session counter. The daemon
|
|
* fixture acquires this value before inspecting attribution. */
|
|
(void)atomic_fetch_add_explicit(&context->session_cancels, 1, memory_order_release);
|
|
if (context->session_cancel_fd >= 0) {
|
|
const char marker = 'C';
|
|
(void)write(context->session_cancel_fd, &marker, 1);
|
|
}
|
|
}
|
|
}
|
|
|
|
static void frontend_eof_application_session_close(
|
|
void *opaque, cbm_daemon_runtime_application_session_t *opaque_session) {
|
|
frontend_eof_application_context_t *context = opaque;
|
|
frontend_eof_application_session_t *session =
|
|
(frontend_eof_application_session_t *)opaque_session;
|
|
if (context && session && session->context == context) {
|
|
free(session);
|
|
}
|
|
}
|
|
|
|
static bool frontend_eof_write_all(int fd, const char *bytes, size_t length) {
|
|
size_t offset = 0;
|
|
while (offset < length) {
|
|
ssize_t written = write(fd, bytes + offset, length - offset);
|
|
if (written > 0) {
|
|
offset += (size_t)written;
|
|
} else if (written < 0 && errno == EINTR) {
|
|
continue;
|
|
} else {
|
|
return false;
|
|
}
|
|
}
|
|
return true;
|
|
}
|
|
|
|
static void *frontend_eof_writer(void *opaque) {
|
|
frontend_eof_writer_t *writer = opaque;
|
|
static const char first[] =
|
|
"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"test/block\",\"params\":{}}\n";
|
|
bool ok = frontend_eof_write_all(writer->fd, first, sizeof(first) - 1);
|
|
uint64_t deadline = cbm_now_ms() + FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS;
|
|
while (
|
|
ok &&
|
|
!atomic_load_explicit(&writer->application->first_request_started, memory_order_acquire) &&
|
|
cbm_now_ms() < deadline) {
|
|
cbm_usleep(1000);
|
|
}
|
|
ok = ok &&
|
|
atomic_load_explicit(&writer->application->first_request_started, memory_order_acquire);
|
|
for (int index = 0; ok && writer->overflow && index < FRONTEND_EOF_TEST_OVERFLOW_MESSAGES;
|
|
index++) {
|
|
char message[160];
|
|
int written =
|
|
snprintf(message, sizeof(message),
|
|
"{\"jsonrpc\":\"2.0\",\"id\":%d,\"method\":\"test/queued\",\"params\":{}}\n",
|
|
index + 2);
|
|
ok = written > 0 && written < (int)sizeof(message) &&
|
|
frontend_eof_write_all(writer->fd, message, (size_t)written);
|
|
}
|
|
ok = close(writer->fd) == 0 && ok;
|
|
writer->fd = -1;
|
|
if (writer->overflow) {
|
|
/* Everything — 32 over-capacity frames AND the EOF behind them — is now
|
|
* committed to the pipe. Only now release the held first request: the
|
|
* queue was provably full while it blocked, so the drain that follows
|
|
* exercises enqueue-wait, not a conveniently fast worker. */
|
|
atomic_store_explicit(&writer->application->release_first_request, true,
|
|
memory_order_release);
|
|
}
|
|
atomic_store_explicit(&writer->succeeded, ok, memory_order_release);
|
|
atomic_store_explicit(&writer->finished, true, memory_order_release);
|
|
return NULL;
|
|
}
|
|
|
|
static const char frontend_recovery_request[] =
|
|
"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/list\",\"params\":{}}\n";
|
|
#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS
|
|
static const char frontend_recovery_cancel[] =
|
|
"{\"jsonrpc\":\"2.0\",\"method\":\"notifications/cancelled\","
|
|
"\"params\":{\"requestId\":1}}\n";
|
|
static const char frontend_recovery_followup[] =
|
|
"{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"tools/list\",\"params\":{}}\n";
|
|
#endif
|
|
|
|
static void *frontend_recovery_writer(void *opaque) {
|
|
frontend_eof_writer_t *writer = opaque;
|
|
bool ok = frontend_eof_write_all(writer->fd, frontend_recovery_request,
|
|
sizeof(frontend_recovery_request) - 1U);
|
|
uint64_t deadline = cbm_now_ms() + FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS;
|
|
if (writer->recovery_mode != FRONTEND_RECOVERY_RETRY) {
|
|
while (ok &&
|
|
!atomic_load_explicit(&writer->application->first_request_started,
|
|
memory_order_acquire) &&
|
|
cbm_now_ms() < deadline) {
|
|
cbm_usleep(1000);
|
|
}
|
|
ok = ok && atomic_load_explicit(&writer->application->first_request_started,
|
|
memory_order_acquire);
|
|
}
|
|
#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS
|
|
if (writer->recovery_mode == FRONTEND_RECOVERY_CANCEL) {
|
|
uint64_t cancellations = cbm_daemon_frontend_test_routed_cancellations();
|
|
ok = ok && frontend_eof_write_all(writer->fd, frontend_recovery_cancel,
|
|
sizeof(frontend_recovery_cancel) - 1U);
|
|
/* Keep context blocked until the reader routes the cancellation. */
|
|
while (ok && cbm_daemon_frontend_test_routed_cancellations() == cancellations &&
|
|
cbm_now_ms() < deadline) {
|
|
cbm_usleep(1000);
|
|
}
|
|
ok = ok && cbm_daemon_frontend_test_routed_cancellations() > cancellations;
|
|
atomic_store_explicit(&writer->application->release_first_request, true,
|
|
memory_order_release);
|
|
ok = ok && frontend_eof_write_all(writer->fd, frontend_recovery_followup,
|
|
sizeof(frontend_recovery_followup) - 1U);
|
|
}
|
|
#endif
|
|
if (writer->recovery_mode == FRONTEND_RECOVERY_CLEAN_EOF) {
|
|
ok = close(writer->fd) == 0 && ok;
|
|
writer->fd = -1;
|
|
cbm_usleep(FRONTEND_RECOVERY_EOF_RELEASE_MS * 1000U);
|
|
atomic_store_explicit(&writer->application->release_first_request, true,
|
|
memory_order_release);
|
|
}
|
|
int expected_requests = writer->recovery_mode == FRONTEND_RECOVERY_CLEAN_EOF ? 1 : 2;
|
|
while (ok &&
|
|
atomic_load_explicit(&writer->application->requests, memory_order_acquire) <
|
|
expected_requests &&
|
|
cbm_now_ms() < deadline) {
|
|
cbm_usleep(1000);
|
|
}
|
|
ok = ok && atomic_load_explicit(&writer->application->requests, memory_order_acquire) ==
|
|
expected_requests;
|
|
if (writer->fd >= 0) {
|
|
ok = close(writer->fd) == 0 && ok;
|
|
writer->fd = -1;
|
|
}
|
|
atomic_store_explicit(&writer->succeeded, ok, memory_order_release);
|
|
atomic_store_explicit(&writer->finished, true, memory_order_release);
|
|
return NULL;
|
|
}
|
|
|
|
static cbm_daemon_runtime_service_t *frontend_eof_service_start(frontend_eof_fixture_t *fixture) {
|
|
cbm_daemon_runtime_application_callbacks_t callbacks = {
|
|
.context = &fixture->application,
|
|
.session_open = frontend_eof_application_session_open,
|
|
.request = frontend_eof_application_request,
|
|
.request_cancel = frontend_eof_application_request_cancel,
|
|
.session_cancel = frontend_eof_application_session_cancel,
|
|
.session_close = frontend_eof_application_session_close,
|
|
};
|
|
cbm_daemon_runtime_service_config_t config = {
|
|
.endpoint = fixture->endpoint,
|
|
.identity = fixture->identity,
|
|
.conflict_log_path = fixture->conflict_log,
|
|
.conflict_log_cap_bytes = 64U * 1024U,
|
|
.max_clients = 4,
|
|
.lease_timeout_ms = 5000,
|
|
.request_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS,
|
|
.shutdown_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS,
|
|
.application = callbacks,
|
|
};
|
|
return cbm_daemon_runtime_service_start(&config);
|
|
}
|
|
|
|
static bool frontend_eof_fixture_start(frontend_eof_fixture_t *fixture, const char *parent) {
|
|
memset(fixture, 0, sizeof(*fixture));
|
|
atomic_init(&fixture->application.requests, 0);
|
|
atomic_init(&fixture->application.session_cancels, 0);
|
|
atomic_init(&fixture->application.sessions_opened, 0);
|
|
atomic_init(&fixture->application.first_session_cancels, 0);
|
|
atomic_init(&fixture->application.second_session_cancels, 0);
|
|
atomic_init(&fixture->application.block_first_request, true);
|
|
atomic_init(&fixture->application.first_request_started, false);
|
|
atomic_init(&fixture->application.release_first_request, false);
|
|
fixture->application.request_observed_fd = -1;
|
|
fixture->application.session_cancel_fd = -1;
|
|
char key[CBM_DAEMON_KEY_SIZE];
|
|
int log_written = snprintf(fixture->conflict_log, sizeof(fixture->conflict_log),
|
|
"%s/conflicts.ndjson", parent);
|
|
if (log_written <= 0 || log_written >= (int)sizeof(fixture->conflict_log) ||
|
|
!cbm_daemon_rendezvous_key(key) ||
|
|
!cbm_daemon_runtime_process_build_fingerprint((uint64_t)getpid(),
|
|
fixture->build_fingerprint)) {
|
|
return false;
|
|
}
|
|
fixture->endpoint = cbm_daemon_ipc_endpoint_new(key, parent);
|
|
fixture->manager = fixture->endpoint ? cbm_version_cohort_manager_new(fixture->endpoint) : NULL;
|
|
fixture->identity = (cbm_daemon_build_identity_t){
|
|
.semantic_version = "2.4.0",
|
|
.build_fingerprint = fixture->build_fingerprint,
|
|
.cache_fingerprint = FRONTEND_TEST_CACHE,
|
|
.protocol_abi = 3,
|
|
.store_abi = 11,
|
|
.feature_abi = 7,
|
|
};
|
|
fixture->service = fixture->manager ? frontend_eof_service_start(fixture) : NULL;
|
|
cbm_daemon_runtime_connect_result_t connect_result = {0};
|
|
fixture->client = fixture->service
|
|
? cbm_daemon_runtime_client_connect(fixture->endpoint, &fixture->identity,
|
|
FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS,
|
|
&connect_result)
|
|
: NULL;
|
|
return fixture->client && connect_result.status == CBM_DAEMON_RUNTIME_CONNECT_ACCEPTED;
|
|
}
|
|
|
|
static bool frontend_eof_fixture_restart_service(frontend_eof_fixture_t *fixture) {
|
|
bool stopped = fixture->service && cbm_daemon_runtime_service_stop(
|
|
fixture->service, FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS);
|
|
bool freed = stopped && cbm_daemon_runtime_service_free(fixture->service);
|
|
fixture->service = freed ? frontend_eof_service_start(fixture) : NULL;
|
|
return stopped && freed && fixture->service;
|
|
}
|
|
|
|
static bool frontend_eof_fixture_finish(frontend_eof_fixture_t *fixture) {
|
|
bool ok = true;
|
|
if (fixture->client) {
|
|
ok = cbm_daemon_runtime_client_close(fixture->client,
|
|
FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS) &&
|
|
ok;
|
|
fixture->client = NULL;
|
|
}
|
|
if (fixture->service) {
|
|
if (cbm_daemon_runtime_service_state(fixture->service) !=
|
|
CBM_DAEMON_RUNTIME_SERVICE_EXITED) {
|
|
ok = cbm_daemon_runtime_service_stop(fixture->service,
|
|
FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS) &&
|
|
ok;
|
|
}
|
|
ok = cbm_daemon_runtime_service_free(fixture->service) && ok;
|
|
fixture->service = NULL;
|
|
}
|
|
if (fixture->manager) {
|
|
ok = cbm_version_cohort_manager_free(&fixture->manager) == CBM_PRIVATE_FILE_LOCK_OK && ok;
|
|
}
|
|
cbm_daemon_ipc_endpoint_free(fixture->endpoint);
|
|
fixture->endpoint = NULL;
|
|
return ok;
|
|
}
|
|
|
|
static int frontend_eof_child_run(const char *parent, bool overflow) {
|
|
frontend_eof_fixture_t fixture;
|
|
if (!frontend_eof_fixture_start(&fixture, parent)) {
|
|
(void)frontend_eof_fixture_finish(&fixture);
|
|
return 70;
|
|
}
|
|
int input_pipe[2] = {-1, -1};
|
|
if (pipe(input_pipe) != 0) {
|
|
(void)frontend_eof_fixture_finish(&fixture);
|
|
return 71;
|
|
}
|
|
FILE *input = fdopen(input_pipe[0], "rb");
|
|
FILE *output = tmpfile();
|
|
frontend_eof_writer_t writer = {
|
|
.fd = input_pipe[1],
|
|
.overflow = overflow,
|
|
.application = &fixture.application,
|
|
};
|
|
atomic_init(&writer.finished, false);
|
|
atomic_init(&writer.succeeded, false);
|
|
cbm_thread_t writer_thread;
|
|
bool writer_started =
|
|
input && output && cbm_thread_create(&writer_thread, 0, frontend_eof_writer, &writer) == 0;
|
|
if (!writer_started) {
|
|
if (input) {
|
|
(void)fclose(input);
|
|
} else {
|
|
(void)close(input_pipe[0]);
|
|
}
|
|
(void)close(input_pipe[1]);
|
|
if (output) {
|
|
(void)fclose(output);
|
|
}
|
|
(void)frontend_eof_fixture_finish(&fixture);
|
|
return 72;
|
|
}
|
|
|
|
cbm_daemon_runtime_client_t *frontend_client = fixture.client;
|
|
fixture.client = NULL; /* cbm_daemon_frontend_mcp_run consumes it. */
|
|
cbm_daemon_frontend_session_config_t session = {
|
|
.bootstrap =
|
|
{
|
|
.role = CBM_DAEMON_PROCESS_MCP_CLIENT,
|
|
.endpoint = fixture.endpoint,
|
|
.identity = &fixture.identity,
|
|
.executable_path = "unused",
|
|
.connect_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS,
|
|
.startup_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS,
|
|
},
|
|
.session_root = parent,
|
|
.tool_profile = CBM_MCP_TOOL_PROFILE_ALL,
|
|
};
|
|
int result =
|
|
cbm_daemon_frontend_mcp_run(frontend_client, fixture.manager, &session, input, output);
|
|
bool joined = cbm_thread_join(&writer_thread) == 0;
|
|
bool writer_ok = atomic_load_explicit(&writer.finished, memory_order_acquire) &&
|
|
atomic_load_explicit(&writer.succeeded, memory_order_acquire);
|
|
bool request_started =
|
|
atomic_load_explicit(&fixture.application.first_request_started, memory_order_acquire);
|
|
bool session_cancelled =
|
|
atomic_load_explicit(&fixture.application.session_cancels, memory_order_acquire) > 0;
|
|
int requests_observed =
|
|
atomic_load_explicit(&fixture.application.requests, memory_order_acquire);
|
|
bool input_closed = fclose(input) == 0;
|
|
bool output_closed = fclose(output) == 0;
|
|
bool streams_closed = input_closed && output_closed;
|
|
bool fixture_closed = frontend_eof_fixture_finish(&fixture);
|
|
bool result_matches_contract;
|
|
if (overflow) {
|
|
/* Over-capacity pipelined input is a client going FAST, not a client
|
|
* going WRONG (#1522 follow-up): enqueue must backpressure the stdin
|
|
* reader until the worker drains, then every queued frame — the held
|
|
* first request plus all 32 over-capacity frames — is answered and the
|
|
* session closes cleanly. The old contract pinned the defect: enqueue
|
|
* overflow failed the whole session (result < 0, session cancelled,
|
|
* every buffered response lost). */
|
|
/* session_cancel also fires on the ordinary client disconnect at EOF
|
|
* teardown, so it cannot discriminate this contract; the loss-free
|
|
* property is the request count plus the clean run result. */
|
|
result_matches_contract =
|
|
result == 0 && requests_observed == 1 + FRONTEND_EOF_TEST_OVERFLOW_MESSAGES;
|
|
} else {
|
|
result_matches_contract = result == 0 && session_cancelled;
|
|
}
|
|
return result_matches_contract && joined && writer_ok && request_started && streams_closed &&
|
|
fixture_closed
|
|
? 0
|
|
: 73;
|
|
}
|
|
|
|
static bool frontend_eof_run_isolated(const char *tag, bool overflow) {
|
|
char parent[FRONTEND_TEST_PATH_CAP];
|
|
int written =
|
|
snprintf(parent, sizeof(parent), "%s/cbm-frontend-eof-%s-XXXXXX", cbm_tmpdir(), tag);
|
|
if (written <= 0 || written >= (int)sizeof(parent) || !cbm_mkdtemp(parent)) {
|
|
return false;
|
|
}
|
|
pid_t child = fork();
|
|
if (child == 0) {
|
|
(void)signal(SIGALRM, SIG_DFL);
|
|
/* This alarm bounds a broken isolated harness, not product behavior.
|
|
* Valid startup, drain/watchdog, close, and fixture teardown can exceed
|
|
* ten seconds under sanitizers while remaining within their own bounds. */
|
|
(void)alarm(FRONTEND_EOF_TEST_CATASTROPHIC_TIMEOUT_S);
|
|
_exit(frontend_eof_child_run(parent, overflow));
|
|
}
|
|
int status = 0;
|
|
pid_t waited;
|
|
do {
|
|
waited = child > 0 ? waitpid(child, &status, 0) : -1;
|
|
} while (waited < 0 && errno == EINTR);
|
|
bool cleaned = th_rmtree(parent) == 0;
|
|
bool child_ok = child > 0 && waited == child && WIFEXITED(status) && WEXITSTATUS(status) == 0;
|
|
if (!child_ok) {
|
|
if (child <= 0) {
|
|
(void)fprintf(stderr, "frontend EOF fixture %s: fork failed\n", tag);
|
|
} else if (waited != child) {
|
|
(void)fprintf(stderr, "frontend EOF fixture %s: waitpid failed errno=%d\n", tag, errno);
|
|
} else if (WIFSIGNALED(status)) {
|
|
(void)fprintf(stderr, "frontend EOF fixture %s: child signal=%d\n", tag,
|
|
WTERMSIG(status));
|
|
} else if (WIFEXITED(status)) {
|
|
(void)fprintf(stderr, "frontend EOF fixture %s: child exit=%d\n", tag,
|
|
WEXITSTATUS(status));
|
|
} else {
|
|
(void)fprintf(stderr, "frontend EOF fixture %s: unexpected child status\n", tag);
|
|
}
|
|
}
|
|
if (!cleaned) {
|
|
(void)fprintf(stderr, "frontend EOF fixture %s: cleanup failed\n", tag);
|
|
}
|
|
return child_ok && cleaned;
|
|
}
|
|
|
|
static int frontend_recovery_child_run(const char *parent, frontend_recovery_mode_t recovery_mode) {
|
|
(void)alarm(FRONTEND_RECOVERY_TEST_CATASTROPHIC_TIMEOUT_S);
|
|
frontend_eof_fixture_t fixture;
|
|
if (!frontend_eof_fixture_start(&fixture, parent)) {
|
|
(void)frontend_eof_fixture_finish(&fixture);
|
|
return 73;
|
|
}
|
|
atomic_store_explicit(&fixture.application.block_first_request,
|
|
recovery_mode != FRONTEND_RECOVERY_RETRY, memory_order_release);
|
|
cbm_daemon_runtime_client_t *stale_client = fixture.client;
|
|
fixture.client = NULL;
|
|
bool replacement_ready = frontend_eof_fixture_restart_service(&fixture);
|
|
int input_pipe[2] = {-1, -1};
|
|
bool pipe_ready = replacement_ready && pipe(input_pipe) == 0;
|
|
FILE *input = pipe_ready ? fdopen(input_pipe[0], "rb") : NULL;
|
|
FILE *output = input ? tmpfile() : NULL;
|
|
frontend_eof_writer_t writer = {
|
|
.fd = input_pipe[1],
|
|
.recovery_mode = recovery_mode,
|
|
.application = &fixture.application,
|
|
};
|
|
atomic_init(&writer.finished, false);
|
|
atomic_init(&writer.succeeded, false);
|
|
cbm_thread_t writer_thread;
|
|
bool writer_started =
|
|
output && cbm_thread_create(&writer_thread, 0, frontend_recovery_writer, &writer) == 0;
|
|
if (!writer_started && input_pipe[1] >= 0) {
|
|
(void)close(input_pipe[1]);
|
|
input_pipe[1] = -1;
|
|
}
|
|
bool request_ready = writer_started;
|
|
cbm_daemon_frontend_session_config_t session = {
|
|
.bootstrap =
|
|
{
|
|
.role = CBM_DAEMON_PROCESS_MCP_CLIENT,
|
|
.endpoint = fixture.endpoint,
|
|
.identity = &fixture.identity,
|
|
.executable_path = "unused",
|
|
.connect_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS,
|
|
.startup_timeout_ms = FRONTEND_RECOVERY_TEST_STARTUP_TIMEOUT_MS,
|
|
},
|
|
.session_root = parent,
|
|
.tool_profile = CBM_MCP_TOOL_PROFILE_ALL,
|
|
};
|
|
int result = request_ready ? cbm_daemon_frontend_mcp_run(stale_client, fixture.manager,
|
|
&session, input, output)
|
|
: -1;
|
|
stale_client = NULL; /* the frontend consumes the stale or replacement handle */
|
|
bool writer_joined = writer_started && cbm_thread_join(&writer_thread) == 0;
|
|
bool writer_ok = writer_joined &&
|
|
atomic_load_explicit(&writer.finished, memory_order_acquire) &&
|
|
atomic_load_explicit(&writer.succeeded, memory_order_acquire);
|
|
char response[512];
|
|
memset(response, 0, sizeof(response));
|
|
bool response_ready = result == 0 && fflush(output) == 0 && fseek(output, 0, SEEK_SET) == 0;
|
|
size_t response_length = response_ready ? fread(response, 1, sizeof(response) - 1U, output) : 0;
|
|
bool response_exact = false;
|
|
if (recovery_mode == FRONTEND_RECOVERY_CANCEL) {
|
|
response_exact = strstr(response, "\"id\":1") && strstr(response, "\"code\":-32800") &&
|
|
strstr(response, frontend_recovery_followup);
|
|
} else if (recovery_mode == FRONTEND_RECOVERY_CLEAN_EOF) {
|
|
response_exact = response_length == 0;
|
|
} else {
|
|
response_exact = response_length == sizeof(frontend_recovery_request) - 1U &&
|
|
memcmp(response, frontend_recovery_request,
|
|
sizeof(frontend_recovery_request) - 1U) == 0;
|
|
}
|
|
int requests = atomic_load_explicit(&fixture.application.requests, memory_order_acquire);
|
|
int expected_requests = recovery_mode == FRONTEND_RECOVERY_CLEAN_EOF ? 1 : 2;
|
|
bool input_closed = input && fclose(input) == 0;
|
|
bool output_closed = output && fclose(output) == 0;
|
|
bool fixture_closed = frontend_eof_fixture_finish(&fixture);
|
|
#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS
|
|
/* The replay restores what startup set within the startup budget. On the
|
|
* connect budget (a probe bound) a replacement daemon that answers slower
|
|
* than a probe would end an otherwise good recovery. Every mode replays. */
|
|
if (cbm_daemon_frontend_test_replay_timeout_ms() != FRONTEND_RECOVERY_TEST_STARTUP_TIMEOUT_MS) {
|
|
return 75;
|
|
}
|
|
#endif
|
|
return replacement_ready && request_ready && result == 0 && response_exact &&
|
|
requests == expected_requests && writer_ok && input_closed && output_closed &&
|
|
fixture_closed
|
|
? 0
|
|
: 74;
|
|
}
|
|
|
|
static bool frontend_recovery_run_isolated(frontend_recovery_mode_t recovery_mode) {
|
|
char parent[FRONTEND_TEST_PATH_CAP];
|
|
int written = snprintf(parent, sizeof(parent), "%s/cbm-frontend-recovery-XXXXXX", cbm_tmpdir());
|
|
if (written <= 0 || written >= (int)sizeof(parent) || !cbm_mkdtemp(parent)) {
|
|
return false;
|
|
}
|
|
pid_t child = fork();
|
|
if (child == 0) {
|
|
(void)signal(SIGALRM, SIG_DFL);
|
|
_exit(frontend_recovery_child_run(parent, recovery_mode));
|
|
}
|
|
int status = 0;
|
|
pid_t waited;
|
|
do {
|
|
waited = child > 0 ? waitpid(child, &status, 0) : -1;
|
|
} while (waited < 0 && errno == EINTR);
|
|
bool cleaned = th_rmtree(parent) == 0;
|
|
bool child_ok = child > 0 && waited == child && WIFEXITED(status) && WEXITSTATUS(status) == 0;
|
|
if (!child_ok && child > 0 && waited == child) {
|
|
(void)fprintf(stderr, "frontend recovery fixture failed: status=0x%x exit=%d signal=%d\n",
|
|
status, WIFEXITED(status) ? WEXITSTATUS(status) : -1,
|
|
WIFSIGNALED(status) ? WTERMSIG(status) : 0);
|
|
}
|
|
return child_ok && cleaned;
|
|
}
|
|
|
|
static void frontend_test_release_lease(cbm_version_cohort_lease_t **lease) {
|
|
while (lease && *lease && cbm_version_cohort_lease_release(lease) != CBM_PRIVATE_FILE_LOCK_OK) {
|
|
cbm_usleep(1000);
|
|
}
|
|
}
|
|
|
|
static bool frontend_test_release_lease_until(cbm_version_cohort_lease_t **lease,
|
|
uint64_t deadline_ms) {
|
|
while (lease && *lease) {
|
|
if (cbm_version_cohort_lease_release(lease) == CBM_PRIVATE_FILE_LOCK_OK) {
|
|
continue;
|
|
}
|
|
if (cbm_now_ms() >= deadline_ms) {
|
|
return false;
|
|
}
|
|
cbm_usleep(1000);
|
|
}
|
|
return !lease || !*lease;
|
|
}
|
|
|
|
static bool frontend_test_read_byte(int fd, char *observed, uint64_t deadline_ms) {
|
|
if (!observed) {
|
|
return false;
|
|
}
|
|
*observed = '\0';
|
|
int flags = fcntl(fd, F_GETFL, 0);
|
|
if (flags < 0 || fcntl(fd, F_SETFL, flags | O_NONBLOCK) != 0) {
|
|
return false;
|
|
}
|
|
for (;;) {
|
|
char value = '\0';
|
|
ssize_t count = read(fd, &value, 1);
|
|
if (count == 1) {
|
|
*observed = value;
|
|
return true;
|
|
}
|
|
if (count == 0 || (count < 0 && errno != EAGAIN && errno != EWOULDBLOCK) ||
|
|
cbm_now_ms() >= deadline_ms) {
|
|
return false;
|
|
}
|
|
cbm_usleep(1000);
|
|
}
|
|
}
|
|
|
|
static bool frontend_test_wait_byte(int fd, char expected, uint64_t deadline_ms) {
|
|
char observed = '\0';
|
|
return frontend_test_read_byte(fd, &observed, deadline_ms) && observed == expected;
|
|
}
|
|
|
|
typedef struct {
|
|
int fd;
|
|
atomic_bool finished;
|
|
atomic_bool succeeded;
|
|
} frontend_backpressure_writer_t;
|
|
|
|
static cbm_version_cohort_quiesce_result_t frontend_test_quiesce_requested(void *context);
|
|
|
|
static bool frontend_backpressure_identity(cbm_daemon_build_identity_t *identity,
|
|
const char build[CBM_DAEMON_BUILD_FINGERPRINT_SIZE]) {
|
|
if (!identity || !build || strlen(build) != CBM_DAEMON_BUILD_FINGERPRINT_SIZE - 1) {
|
|
return false;
|
|
}
|
|
*identity = (cbm_daemon_build_identity_t){
|
|
.semantic_version = "2.4.0",
|
|
.build_fingerprint = build,
|
|
.cache_fingerprint = FRONTEND_TEST_CACHE,
|
|
.protocol_abi = 3,
|
|
.store_abi = 11,
|
|
.feature_abi = 7,
|
|
};
|
|
return true;
|
|
}
|
|
|
|
static void *frontend_backpressure_writer(void *opaque) {
|
|
frontend_backpressure_writer_t *writer = opaque;
|
|
static const char prefix[] = "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"test/large\","
|
|
"\"params\":{\"blob\":\"";
|
|
static const char suffix[] = "\"}}\n";
|
|
char *message = malloc(FRONTEND_BACKPRESSURE_MESSAGE_BYTES);
|
|
size_t prefix_length = sizeof(prefix) - 1;
|
|
size_t suffix_length = sizeof(suffix) - 1;
|
|
bool valid = message && prefix_length + suffix_length < FRONTEND_BACKPRESSURE_MESSAGE_BYTES;
|
|
if (valid) {
|
|
memcpy(message, prefix, prefix_length);
|
|
memset(message + prefix_length, 'x',
|
|
FRONTEND_BACKPRESSURE_MESSAGE_BYTES - prefix_length - suffix_length);
|
|
memcpy(message + FRONTEND_BACKPRESSURE_MESSAGE_BYTES - suffix_length, suffix,
|
|
suffix_length);
|
|
valid = frontend_eof_write_all(writer->fd, message, FRONTEND_BACKPRESSURE_MESSAGE_BYTES);
|
|
}
|
|
free(message);
|
|
valid = close(writer->fd) == 0 && valid;
|
|
writer->fd = -1;
|
|
atomic_store_explicit(&writer->succeeded, valid, memory_order_release);
|
|
atomic_store_explicit(&writer->finished, true, memory_order_release);
|
|
return NULL;
|
|
}
|
|
|
|
static int frontend_backpressure_daemon_run(const char *parent, int ready_fd, int cancel_fd,
|
|
int control_fd,
|
|
const char build[CBM_DAEMON_BUILD_FINGERPRINT_SIZE]) {
|
|
(void)alarm(FRONTEND_BACKPRESSURE_DAEMON_TIMEOUT_S);
|
|
char key[CBM_DAEMON_KEY_SIZE];
|
|
char conflict_log[FRONTEND_TEST_PATH_CAP];
|
|
cbm_daemon_build_identity_t identity = {0};
|
|
int log_written = snprintf(conflict_log, sizeof(conflict_log), "%s/conflicts.ndjson", parent);
|
|
cbm_daemon_ipc_endpoint_t *endpoint =
|
|
log_written > 0 && log_written < (int)sizeof(conflict_log) &&
|
|
cbm_daemon_rendezvous_key(key) && frontend_backpressure_identity(&identity, build)
|
|
? cbm_daemon_ipc_endpoint_new(key, parent)
|
|
: NULL;
|
|
frontend_eof_application_context_t application;
|
|
memset(&application, 0, sizeof(application));
|
|
atomic_init(&application.requests, 0);
|
|
atomic_init(&application.session_cancels, 0);
|
|
atomic_init(&application.sessions_opened, 0);
|
|
atomic_init(&application.first_session_cancels, 0);
|
|
atomic_init(&application.second_session_cancels, 0);
|
|
atomic_init(&application.block_first_request, false);
|
|
atomic_init(&application.first_request_started, false);
|
|
atomic_init(&application.release_first_request, false);
|
|
application.request_observed_fd = ready_fd;
|
|
application.session_cancel_fd = cancel_fd;
|
|
cbm_daemon_runtime_application_callbacks_t callbacks = {
|
|
.context = &application,
|
|
.session_open = frontend_eof_application_session_open,
|
|
.request = frontend_eof_application_request,
|
|
.request_cancel = frontend_eof_application_request_cancel,
|
|
.session_cancel = frontend_eof_application_session_cancel,
|
|
.session_close = frontend_eof_application_session_close,
|
|
};
|
|
cbm_daemon_runtime_service_config_t config = {
|
|
.endpoint = endpoint,
|
|
.identity = identity,
|
|
.conflict_log_path = conflict_log,
|
|
.conflict_log_cap_bytes = 64U * 1024U,
|
|
.max_clients = 4,
|
|
.lease_timeout_ms = 30000,
|
|
.request_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS,
|
|
.shutdown_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS,
|
|
.application = callbacks,
|
|
};
|
|
cbm_daemon_runtime_service_t *service =
|
|
endpoint ? cbm_daemon_runtime_service_start(&config) : NULL;
|
|
const char ready = 'R';
|
|
bool announced = service && write(ready_fd, &ready, 1) == 1;
|
|
uint64_t cancellation_deadline = cbm_now_ms() + FRONTEND_BACKPRESSURE_RUNTIME_TIMEOUT_MS;
|
|
while (announced &&
|
|
atomic_load_explicit(&application.session_cancels, memory_order_acquire) == 0 &&
|
|
cbm_now_ms() < cancellation_deadline) {
|
|
cbm_usleep(1000);
|
|
}
|
|
int target_cancellations =
|
|
atomic_load_explicit(&application.session_cancels, memory_order_acquire);
|
|
int sentinel_cancellations_before_close =
|
|
atomic_load_explicit(&application.first_session_cancels, memory_order_acquire);
|
|
int target_session_cancellations =
|
|
atomic_load_explicit(&application.second_session_cancels, memory_order_acquire);
|
|
bool target_cancelled_exactly_once = announced && target_cancellations == 1 &&
|
|
sentinel_cancellations_before_close == 0 &&
|
|
target_session_cancellations == 1;
|
|
const char checkpoint = target_cancelled_exactly_once ? 'T' : 'E';
|
|
bool checkpoint_written = write(ready_fd, &checkpoint, 1) == 1;
|
|
char control = '\0';
|
|
bool shutdown_released =
|
|
checkpoint_written &&
|
|
frontend_test_read_byte(control_fd, &control,
|
|
cbm_now_ms() + FRONTEND_BACKPRESSURE_RUNTIME_TIMEOUT_MS) &&
|
|
control == 'S';
|
|
int final_cancellations =
|
|
atomic_load_explicit(&application.session_cancels, memory_order_acquire);
|
|
int sentinel_session_cancellations =
|
|
atomic_load_explicit(&application.first_session_cancels, memory_order_acquire);
|
|
int final_target_session_cancellations =
|
|
atomic_load_explicit(&application.second_session_cancels, memory_order_acquire);
|
|
bool sentinel_closed_exactly_once = final_cancellations == 2 &&
|
|
sentinel_session_cancellations == 1 &&
|
|
final_target_session_cancellations == 1;
|
|
bool stopped =
|
|
service && cbm_daemon_runtime_service_stop(service, FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS);
|
|
bool exited = stopped && cbm_daemon_runtime_service_wait_exited(service, 0);
|
|
bool freed = exited && cbm_daemon_runtime_service_free(service);
|
|
cbm_daemon_ipc_endpoint_free(endpoint);
|
|
const char done = target_cancelled_exactly_once && shutdown_released &&
|
|
sentinel_closed_exactly_once && stopped && exited && freed
|
|
? 'D'
|
|
: 'E';
|
|
bool finished = write(ready_fd, &done, 1) == 1;
|
|
(void)close(ready_fd);
|
|
(void)close(cancel_fd);
|
|
(void)close(control_fd);
|
|
return announced && target_cancelled_exactly_once && checkpoint_written && shutdown_released &&
|
|
sentinel_closed_exactly_once && stopped && exited && freed && finished
|
|
? 0
|
|
: 80;
|
|
}
|
|
|
|
static int frontend_backpressure_frontend_run(const char *parent, int input_fd, int input_write_fd,
|
|
int output_fd,
|
|
const char build[CBM_DAEMON_BUILD_FINGERPRINT_SIZE],
|
|
bool keep_input_open) {
|
|
(void)alarm(FRONTEND_BACKPRESSURE_FRONTEND_TIMEOUT_S);
|
|
char key[CBM_DAEMON_KEY_SIZE];
|
|
cbm_daemon_build_identity_t identity = {0};
|
|
cbm_daemon_ipc_endpoint_t *endpoint =
|
|
cbm_daemon_rendezvous_key(key) && frontend_backpressure_identity(&identity, build)
|
|
? cbm_daemon_ipc_endpoint_new(key, parent)
|
|
: NULL;
|
|
cbm_version_cohort_manager_t *manager =
|
|
endpoint ? cbm_version_cohort_manager_new(endpoint) : NULL;
|
|
cbm_version_cohort_lease_t *participant = NULL;
|
|
cbm_daemon_conflict_t conflict;
|
|
cbm_version_cohort_status_t admitted =
|
|
manager ? cbm_version_cohort_acquire(manager, &identity, cbm_now_ms() + 2000U, &participant,
|
|
&conflict)
|
|
: CBM_VERSION_COHORT_IO;
|
|
cbm_daemon_runtime_connect_result_t connect_result = {0};
|
|
cbm_daemon_runtime_client_t *client =
|
|
admitted == CBM_VERSION_COHORT_OK
|
|
? cbm_daemon_runtime_client_connect(
|
|
endpoint, &identity, FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS, &connect_result)
|
|
: NULL;
|
|
FILE *input = client ? fdopen(input_fd, "rb") : NULL;
|
|
FILE *output = input ? fdopen(output_fd, "wb") : NULL;
|
|
int input_hold_fd = keep_input_open && output ? dup(input_write_fd) : -1;
|
|
frontend_backpressure_writer_t writer = {
|
|
.fd = input_write_fd,
|
|
};
|
|
atomic_init(&writer.finished, false);
|
|
atomic_init(&writer.succeeded, false);
|
|
cbm_thread_t writer_thread;
|
|
bool writer_started =
|
|
output && (!keep_input_open || input_hold_fd >= 0) &&
|
|
cbm_thread_create(&writer_thread, 0, frontend_backpressure_writer, &writer) == 0;
|
|
if (!writer_started) {
|
|
if (client) {
|
|
(void)cbm_daemon_runtime_client_close(client, FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS);
|
|
}
|
|
if (input) {
|
|
(void)fclose(input);
|
|
} else {
|
|
(void)close(input_fd);
|
|
}
|
|
if (output) {
|
|
(void)fclose(output);
|
|
} else {
|
|
(void)close(output_fd);
|
|
}
|
|
(void)close(input_write_fd);
|
|
if (input_hold_fd >= 0) {
|
|
(void)close(input_hold_fd);
|
|
}
|
|
frontend_test_release_lease(&participant);
|
|
if (manager) {
|
|
(void)cbm_version_cohort_manager_free(&manager);
|
|
}
|
|
cbm_daemon_ipc_endpoint_free(endpoint);
|
|
return 81;
|
|
}
|
|
|
|
cbm_daemon_frontend_session_config_t session = {
|
|
.bootstrap =
|
|
{
|
|
.role = CBM_DAEMON_PROCESS_MCP_CLIENT,
|
|
.endpoint = endpoint,
|
|
.identity = &identity,
|
|
.executable_path = "unused",
|
|
.connect_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS,
|
|
.startup_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS,
|
|
},
|
|
.session_root = parent,
|
|
.tool_profile = CBM_MCP_TOOL_PROFILE_ALL,
|
|
};
|
|
int result = cbm_daemon_frontend_mcp_run(client, manager, &session, input, output);
|
|
bool joined = cbm_thread_join(&writer_thread) == 0;
|
|
bool writer_ok = atomic_load_explicit(&writer.finished, memory_order_acquire) &&
|
|
atomic_load_explicit(&writer.succeeded, memory_order_acquire);
|
|
bool input_closed = fclose(input) == 0;
|
|
bool output_closed = fclose(output) == 0;
|
|
bool input_hold_closed = input_hold_fd < 0 || close(input_hold_fd) == 0;
|
|
frontend_test_release_lease(&participant);
|
|
bool manager_closed = cbm_version_cohort_manager_free(&manager) == CBM_PRIVATE_FILE_LOCK_OK;
|
|
cbm_daemon_ipc_endpoint_free(endpoint);
|
|
return result < 0 && joined && writer_ok && input_closed && output_closed &&
|
|
input_hold_closed && manager_closed
|
|
? 47
|
|
: 82;
|
|
}
|
|
|
|
static bool frontend_backpressure_run_isolated(bool maintenance) {
|
|
char parent[FRONTEND_TEST_PATH_CAP];
|
|
char build[CBM_DAEMON_BUILD_FINGERPRINT_SIZE] = {0};
|
|
int path_written = snprintf(parent, sizeof(parent),
|
|
maintenance ? "%s/cbm-frontend-maintenance-backpressure-XXXXXX"
|
|
: "%s/cbm-frontend-backpressure-XXXXXX",
|
|
cbm_tmpdir());
|
|
int ready_pipe[2] = {-1, -1};
|
|
int cancel_pipe[2] = {-1, -1};
|
|
int control_pipe[2] = {-1, -1};
|
|
bool directory_ready =
|
|
path_written > 0 && path_written < (int)sizeof(parent) && cbm_mkdtemp(parent);
|
|
bool identity_ready =
|
|
directory_ready && cbm_daemon_runtime_process_build_fingerprint((uint64_t)getpid(), build);
|
|
bool prepared = identity_ready && pipe(ready_pipe) == 0 && pipe(cancel_pipe) == 0 &&
|
|
pipe(control_pipe) == 0;
|
|
if (!prepared) {
|
|
for (size_t index = 0; index < 2; index++) {
|
|
if (ready_pipe[index] >= 0) {
|
|
(void)close(ready_pipe[index]);
|
|
}
|
|
if (cancel_pipe[index] >= 0) {
|
|
(void)close(cancel_pipe[index]);
|
|
}
|
|
if (control_pipe[index] >= 0) {
|
|
(void)close(control_pipe[index]);
|
|
}
|
|
}
|
|
if (directory_ready) {
|
|
(void)th_rmtree(parent);
|
|
}
|
|
return false;
|
|
}
|
|
pid_t daemon = prepared ? fork() : -1;
|
|
if (daemon == 0) {
|
|
(void)signal(SIGALRM, SIG_DFL);
|
|
(void)close(ready_pipe[0]);
|
|
(void)close(cancel_pipe[0]);
|
|
(void)close(control_pipe[1]);
|
|
_exit(frontend_backpressure_daemon_run(parent, ready_pipe[1], cancel_pipe[1],
|
|
control_pipe[0], build));
|
|
}
|
|
if (prepared) {
|
|
(void)close(ready_pipe[1]);
|
|
(void)close(cancel_pipe[1]);
|
|
(void)close(control_pipe[0]);
|
|
}
|
|
char announced_marker = '\0';
|
|
/* The daemon child runs cbm_daemon_runtime_service_start before it can
|
|
* announce, and that includes the cohort-claim path whose own internal
|
|
* budget is HOST_DAEMON_CLAIM_TIMEOUT_MS (12s). An announce deadline
|
|
* below that budget SIGKILLs a slow-but-legitimate startup — on
|
|
* oversubscribed CI runners this fired every round (announced=0,
|
|
* daemon_signal=9) while fast local machines never saw it. Keep the
|
|
* deadline comfortably above the daemon's own worst-case budget.
|
|
*
|
|
* This wait is a LIVENESS BACKSTOP, not a race budget: the daemon either
|
|
* announces or it does not, and the test asserts the announcement, never a
|
|
* timing window. A backstop is only doing its job if it never fires on a
|
|
* legitimately slow start — so it has to scale with the build. Under
|
|
* MemorySanitizer (instrumented libc++ and origin tracking) startup runs
|
|
* several times slower than the 12s budget it must clear, which is what
|
|
* reddened test-msan while every other lane and the same suite under local
|
|
* MSan stayed green. Widening the backstop for sanitized builds costs
|
|
* nothing when the daemon is healthy: a passing run returns as soon as the
|
|
* byte arrives, whatever the ceiling is. */
|
|
#if CBM_SANITIZED
|
|
const uint64_t announce_backstop_ms = 180000U;
|
|
#else
|
|
const uint64_t announce_backstop_ms = 30000U;
|
|
#endif
|
|
bool announced_read =
|
|
daemon > 0 && frontend_test_read_byte(ready_pipe[0], &announced_marker,
|
|
cbm_now_ms() + announce_backstop_ms);
|
|
bool announced = announced_read && announced_marker == 'R';
|
|
|
|
/* This raw runtime client intentionally owns no cohort lease. It is a
|
|
* sentinel for session isolation: maintenance must remove the target
|
|
* frontend without touching this independently authenticated session. Its
|
|
* completed admission precedes the target fork, so the daemon fixture can
|
|
* attribute session-cancel callbacks to sentinel index 0 and target index
|
|
* 1 rather than relying on a global callback count. */
|
|
char sentinel_key[CBM_DAEMON_KEY_SIZE];
|
|
cbm_daemon_build_identity_t sentinel_identity = {0};
|
|
cbm_daemon_ipc_endpoint_t *sentinel_endpoint =
|
|
announced && cbm_daemon_rendezvous_key(sentinel_key) &&
|
|
frontend_backpressure_identity(&sentinel_identity, build)
|
|
? cbm_daemon_ipc_endpoint_new(sentinel_key, parent)
|
|
: NULL;
|
|
cbm_daemon_runtime_connect_result_t sentinel_connect = {0};
|
|
cbm_daemon_runtime_client_t *sentinel =
|
|
sentinel_endpoint ? cbm_daemon_runtime_client_connect(sentinel_endpoint, &sentinel_identity,
|
|
FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS,
|
|
&sentinel_connect)
|
|
: NULL;
|
|
bool sentinel_ready =
|
|
sentinel && sentinel_connect.status == CBM_DAEMON_RUNTIME_CONNECT_ACCEPTED;
|
|
if (!sentinel_ready && daemon > 0) {
|
|
(void)kill(daemon, SIGKILL);
|
|
}
|
|
|
|
int input_pipe[2] = {-1, -1};
|
|
int output_pipe[2] = {-1, -1};
|
|
bool input_ready = sentinel_ready && pipe(input_pipe) == 0;
|
|
bool frontend_pipes = input_ready && pipe(output_pipe) == 0;
|
|
if (input_ready && !frontend_pipes) {
|
|
(void)close(input_pipe[0]);
|
|
(void)close(input_pipe[1]);
|
|
input_pipe[0] = -1;
|
|
input_pipe[1] = -1;
|
|
}
|
|
pid_t frontend = frontend_pipes ? fork() : -1;
|
|
if (frontend == 0) {
|
|
(void)signal(SIGALRM, SIG_DFL);
|
|
(void)close(output_pipe[0]);
|
|
(void)close(ready_pipe[0]);
|
|
(void)close(cancel_pipe[0]);
|
|
(void)close(control_pipe[1]);
|
|
_exit(frontend_backpressure_frontend_run(parent, input_pipe[0], input_pipe[1],
|
|
output_pipe[1], build, maintenance));
|
|
}
|
|
if (frontend_pipes) {
|
|
(void)close(input_pipe[0]);
|
|
(void)close(input_pipe[1]);
|
|
(void)close(output_pipe[1]);
|
|
}
|
|
|
|
char request_marker = '\0';
|
|
bool request_read = frontend > 0 && frontend_test_read_byte(ready_pipe[0], &request_marker,
|
|
cbm_now_ms() + 4000U);
|
|
bool request_observed = request_read && request_marker == 'Q';
|
|
char output_marker = '\0';
|
|
bool output_observed =
|
|
!maintenance || (request_observed && frontend_test_read_byte(output_pipe[0], &output_marker,
|
|
cbm_now_ms() + 4000U));
|
|
|
|
cbm_daemon_ipc_endpoint_t *mutation_endpoint = NULL;
|
|
cbm_version_cohort_manager_t *mutation_manager = NULL;
|
|
cbm_version_cohort_lease_t *mutation = NULL;
|
|
cbm_version_cohort_quiesce_result_t quiesce = CBM_VERSION_COHORT_QUIESCE_NOT_NEEDED;
|
|
cbm_version_cohort_status_t mutation_status = CBM_VERSION_COHORT_OK;
|
|
if (maintenance) {
|
|
char mutation_key[CBM_DAEMON_KEY_SIZE];
|
|
mutation_endpoint = output_observed && cbm_daemon_rendezvous_key(mutation_key)
|
|
? cbm_daemon_ipc_endpoint_new(mutation_key, parent)
|
|
: NULL;
|
|
mutation_manager =
|
|
mutation_endpoint ? cbm_version_cohort_manager_new(mutation_endpoint) : NULL;
|
|
mutation_status = mutation_manager
|
|
? cbm_version_cohort_reserve_for_mutation(
|
|
mutation_manager, cbm_now_ms() + 8000U,
|
|
frontend_test_quiesce_requested, NULL, &quiesce, &mutation)
|
|
: CBM_VERSION_COHORT_IO;
|
|
if (mutation_status != CBM_VERSION_COHORT_OK && frontend > 0) {
|
|
(void)kill(frontend, SIGKILL);
|
|
}
|
|
}
|
|
|
|
int frontend_status = 0;
|
|
pid_t frontend_waited;
|
|
if (frontend > 0) {
|
|
do {
|
|
frontend_waited = waitpid(frontend, &frontend_status, 0);
|
|
} while (frontend_waited < 0 && errno == EINTR);
|
|
} else {
|
|
frontend_waited = -1;
|
|
}
|
|
char cancel_marker = '\0';
|
|
bool cancel_read = request_observed && frontend_test_read_byte(cancel_pipe[0], &cancel_marker,
|
|
cbm_now_ms() + 4000U);
|
|
bool cancelled = cancel_read && cancel_marker == 'C';
|
|
char checkpoint_marker = '\0';
|
|
bool checkpoint_read =
|
|
frontend_test_read_byte(ready_pipe[0], &checkpoint_marker, cbm_now_ms() + 4000U);
|
|
bool target_cancelled_exactly_once = checkpoint_read && checkpoint_marker == 'T';
|
|
bool sentinel_usable =
|
|
cancelled && target_cancelled_exactly_once && sentinel_ready &&
|
|
cbm_daemon_runtime_client_heartbeat(sentinel, FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS);
|
|
bool sentinel_closed =
|
|
sentinel && cbm_daemon_runtime_client_close(sentinel, FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS);
|
|
sentinel = NULL;
|
|
cbm_daemon_ipc_endpoint_free(sentinel_endpoint);
|
|
sentinel_endpoint = NULL;
|
|
const char shutdown = 'S';
|
|
bool shutdown_sent = checkpoint_read && write(control_pipe[1], &shutdown, 1) == 1;
|
|
if (control_pipe[1] >= 0) {
|
|
(void)close(control_pipe[1]);
|
|
control_pipe[1] = -1;
|
|
}
|
|
char done_marker = '\0';
|
|
bool done_read =
|
|
shutdown_sent && frontend_test_read_byte(ready_pipe[0], &done_marker, cbm_now_ms() + 4000U);
|
|
bool daemon_done = done_read && done_marker == 'D';
|
|
if (!daemon_done && daemon > 0) {
|
|
(void)kill(daemon, SIGKILL);
|
|
}
|
|
int daemon_status = 0;
|
|
pid_t daemon_waited;
|
|
do {
|
|
daemon_waited = daemon > 0 ? waitpid(daemon, &daemon_status, 0) : -1;
|
|
} while (daemon_waited < 0 && errno == EINTR);
|
|
|
|
uint64_t cleanup_now = cbm_now_ms();
|
|
uint64_t cleanup_deadline = cleanup_now > UINT64_MAX - FRONTEND_BACKPRESSURE_CLEANUP_TIMEOUT_MS
|
|
? UINT64_MAX
|
|
: cleanup_now + FRONTEND_BACKPRESSURE_CLEANUP_TIMEOUT_MS;
|
|
bool mutation_released = frontend_test_release_lease_until(&mutation, cleanup_deadline);
|
|
while (mutation_released && mutation_manager && cbm_now_ms() < cleanup_deadline &&
|
|
cbm_version_cohort_manager_free(&mutation_manager) != CBM_PRIVATE_FILE_LOCK_OK) {
|
|
cbm_usleep(1000);
|
|
}
|
|
bool mutation_manager_closed = mutation_manager == NULL;
|
|
cbm_daemon_ipc_endpoint_free(mutation_endpoint);
|
|
|
|
if (output_pipe[0] >= 0) {
|
|
(void)close(output_pipe[0]);
|
|
}
|
|
if (ready_pipe[0] >= 0) {
|
|
(void)close(ready_pipe[0]);
|
|
}
|
|
if (cancel_pipe[0] >= 0) {
|
|
(void)close(cancel_pipe[0]);
|
|
}
|
|
if (control_pipe[1] >= 0) {
|
|
(void)close(control_pipe[1]);
|
|
}
|
|
bool cleaned = prepared && th_rmtree(parent) == 0;
|
|
bool frontend_exited_boundedly = frontend_waited == frontend && WIFEXITED(frontend_status) &&
|
|
(maintenance ? WEXITSTATUS(frontend_status) == EXIT_SUCCESS
|
|
: (WEXITSTATUS(frontend_status) == EXIT_FAILURE ||
|
|
WEXITSTATUS(frontend_status) == 47));
|
|
bool daemon_clean =
|
|
daemon_waited == daemon && WIFEXITED(daemon_status) && WEXITSTATUS(daemon_status) == 0;
|
|
bool maintenance_completed =
|
|
!maintenance || (output_observed && mutation_status == CBM_VERSION_COHORT_OK &&
|
|
quiesce == CBM_VERSION_COHORT_QUIESCE_REQUESTED);
|
|
bool passed = prepared && announced && sentinel_ready && frontend_pipes && frontend > 0 &&
|
|
request_observed && maintenance_completed && frontend_exited_boundedly &&
|
|
cancelled && target_cancelled_exactly_once && sentinel_usable &&
|
|
sentinel_closed && shutdown_sent && daemon_done && daemon_clean &&
|
|
mutation_released && mutation_manager_closed && cleaned;
|
|
if (!passed) {
|
|
bool frontend_reaped = frontend > 0 && frontend_waited == frontend;
|
|
bool daemon_reaped = daemon > 0 && daemon_waited == daemon;
|
|
int frontend_exit =
|
|
frontend_reaped && WIFEXITED(frontend_status) ? WEXITSTATUS(frontend_status) : -1;
|
|
int frontend_signal =
|
|
frontend_reaped && WIFSIGNALED(frontend_status) ? WTERMSIG(frontend_status) : 0;
|
|
int daemon_exit =
|
|
daemon_reaped && WIFEXITED(daemon_status) ? WEXITSTATUS(daemon_status) : -1;
|
|
int daemon_signal =
|
|
daemon_reaped && WIFSIGNALED(daemon_status) ? WTERMSIG(daemon_status) : 0;
|
|
fprintf(stderr,
|
|
"frontend backpressure fixture failed: maintenance=%d prepared=%d daemon_pid=%ld "
|
|
"announced=%d announced_marker=0x%02x sentinel_ready=%d frontend_pipes=%d "
|
|
"frontend_pid=%ld frontend_reaped=%d frontend_status=0x%x "
|
|
"frontend_exit=%d frontend_signal=%d request_read=%d "
|
|
"request_marker=0x%02x output_observed=%d output_marker=0x%02x "
|
|
"mutation_status=%d quiesce=%d cancel_read=%d cancel_marker=0x%02x "
|
|
"checkpoint_read=%d checkpoint_marker=0x%02x sentinel_usable=%d "
|
|
"sentinel_closed=%d shutdown_sent=%d "
|
|
"done_read=%d done_marker=0x%02x daemon_reaped=%d "
|
|
"daemon_status=0x%x daemon_exit=%d daemon_signal=%d "
|
|
"mutation_released=%d mutation_manager_closed=%d cleaned=%d\n",
|
|
maintenance, prepared, (long)daemon, announced,
|
|
(unsigned int)(unsigned char)announced_marker, sentinel_ready, frontend_pipes,
|
|
(long)frontend, frontend_reaped, frontend_status, frontend_exit, frontend_signal,
|
|
request_read, (unsigned int)(unsigned char)request_marker, output_observed,
|
|
(unsigned int)(unsigned char)output_marker, mutation_status, quiesce, cancel_read,
|
|
(unsigned int)(unsigned char)cancel_marker, checkpoint_read,
|
|
(unsigned int)(unsigned char)checkpoint_marker, sentinel_usable, sentinel_closed,
|
|
shutdown_sent, done_read, (unsigned int)(unsigned char)done_marker, daemon_reaped,
|
|
daemon_status, daemon_exit, daemon_signal, mutation_released,
|
|
mutation_manager_closed, cleaned);
|
|
}
|
|
return passed;
|
|
}
|
|
|
|
static cbm_version_cohort_quiesce_result_t frontend_test_quiesce_requested(void *context) {
|
|
(void)context;
|
|
return CBM_VERSION_COHORT_QUIESCE_REQUESTED;
|
|
}
|
|
|
|
static bool frontend_test_cancel_active(void *context) {
|
|
int fd = *(int *)context;
|
|
const char marker = 'C';
|
|
return write(fd, &marker, 1) == 1;
|
|
}
|
|
|
|
static bool frontend_test_cancel_and_mark(void *context) {
|
|
atomic_bool *cancelled = context;
|
|
atomic_store_explicit(cancelled, true, memory_order_release);
|
|
return true;
|
|
}
|
|
#endif
|
|
|
|
TEST(daemon_frontend_recognizes_exact_cancellation_notification) {
|
|
ASSERT_TRUE(cbm_daemon_frontend_is_cancellation_notification(
|
|
"{\"jsonrpc\":\"2.0\",\"method\":\"notifications/cancelled\","
|
|
"\"params\":{\"requestId\":7}}"));
|
|
ASSERT_TRUE(cbm_daemon_frontend_is_cancellation_notification(
|
|
"{ \"params\": {}, \"method\": \"notifications/cancelled\", "
|
|
"\"jsonrpc\": \"2.0\" }"));
|
|
PASS();
|
|
}
|
|
|
|
/* RED on the whole-session cancellation shortcut: merely recognizing the
|
|
* notification method is not authority to close a session. The requestId
|
|
* must match the exact numeric/string identity currently being executed. */
|
|
TEST(daemon_frontend_correlates_cancellation_to_exact_request) {
|
|
ASSERT_TRUE(cbm_daemon_frontend_cancellation_matches_request(
|
|
"{\"jsonrpc\":\"2.0\",\"method\":\"notifications/cancelled\","
|
|
"\"params\":{\"requestId\":7}}",
|
|
7, NULL));
|
|
ASSERT_FALSE(cbm_daemon_frontend_cancellation_matches_request(
|
|
"{\"jsonrpc\":\"2.0\",\"method\":\"notifications/cancelled\","
|
|
"\"params\":{\"requestId\":8}}",
|
|
7, NULL));
|
|
ASSERT_TRUE(cbm_daemon_frontend_cancellation_matches_request(
|
|
"{\"jsonrpc\":\"2.0\",\"method\":\"notifications/cancelled\","
|
|
"\"params\":{\"requestId\":\"request-7\"}}",
|
|
-1, "request-7"));
|
|
ASSERT_FALSE(cbm_daemon_frontend_cancellation_matches_request(
|
|
"{\"jsonrpc\":\"2.0\",\"method\":\"notifications/cancelled\","
|
|
"\"params\":{\"requestId\":7}}",
|
|
-1, "7"));
|
|
ASSERT_FALSE(cbm_daemon_frontend_cancellation_matches_request(
|
|
"{\"jsonrpc\":\"2.0\",\"method\":\"notifications/cancelled\","
|
|
"\"params\":{}}",
|
|
7, NULL));
|
|
ASSERT_FALSE(
|
|
cbm_daemon_frontend_cancellation_matches_request("{\"jsonrpc\":\"2.0\",\"id\":9,"
|
|
"\"method\":\"notifications/cancelled\","
|
|
"\"params\":{\"requestId\":7}}",
|
|
7, NULL));
|
|
PASS();
|
|
}
|
|
|
|
/* #2144 (a): the -32800 reply to a cancelled index_repository call is the one
|
|
* text a deadline-bound client can still read, so it names the async mode;
|
|
* other requests and unparseable input keep the plain message. */
|
|
TEST(daemon_frontend_cancelled_index_reply_offers_async_issue2144) {
|
|
const char *index_reply = cbm_daemon_frontend_cancelled_error_message(
|
|
"{\"jsonrpc\":\"2.0\",\"id\":5,\"method\":\"tools/call\","
|
|
"\"params\":{\"name\":\"index_repository\",\"arguments\":{\"repo_path\":\"/r\"}}}");
|
|
ASSERT_NOT_NULL(strstr(index_reply, "Request cancelled"));
|
|
ASSERT_NOT_NULL(strstr(index_reply, "async: true"));
|
|
ASSERT_NOT_NULL(strstr(index_reply, "status: true"));
|
|
ASSERT_NULL(strchr(index_reply, '"'));
|
|
ASSERT_NULL(strchr(index_reply, '\\'));
|
|
ASSERT_STR_EQ(cbm_daemon_frontend_cancelled_error_message(
|
|
"{\"jsonrpc\":\"2.0\",\"id\":6,\"method\":\"tools/call\","
|
|
"\"params\":{\"name\":\"search_graph\",\"arguments\":{}}}"),
|
|
"Request cancelled");
|
|
ASSERT_STR_EQ(cbm_daemon_frontend_cancelled_error_message("not json"), "Request cancelled");
|
|
ASSERT_STR_EQ(cbm_daemon_frontend_cancelled_error_message(NULL), "Request cancelled");
|
|
PASS();
|
|
}
|
|
|
|
TEST(daemon_frontend_ignores_cancellation_text_in_string_content) {
|
|
ASSERT_FALSE(cbm_daemon_frontend_is_cancellation_notification(
|
|
"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\","
|
|
"\"params\":{\"name\":\"search_graph\",\"arguments\":{\"query\":"
|
|
"\"notifications/cancelled\"}}}"));
|
|
ASSERT_FALSE(cbm_daemon_frontend_is_cancellation_notification(
|
|
"{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"tools/call\","
|
|
"\"params\":{\"method\":\"notifications/cancelled\"}}"));
|
|
PASS();
|
|
}
|
|
|
|
TEST(daemon_frontend_rejects_non_notification_cancellation_shapes) {
|
|
ASSERT_FALSE(cbm_daemon_frontend_is_cancellation_notification(
|
|
"{\"jsonrpc\":\"2.0\",\"id\":3,"
|
|
"\"method\":\"notifications/cancelled\"}"));
|
|
ASSERT_FALSE(cbm_daemon_frontend_is_cancellation_notification(
|
|
"{\"jsonrpc\":\"2.0\",\"method\":\"notifications/cancelled-extra\"}"));
|
|
ASSERT_FALSE(cbm_daemon_frontend_is_cancellation_notification(
|
|
"{\"jsonrpc\":\"2.0\",\"method\":\"prefix/notifications/cancelled\"}"));
|
|
ASSERT_FALSE(cbm_daemon_frontend_is_cancellation_notification(
|
|
"{\"jsonrpc\":\"2.0\",\"params\":{\"method\":"
|
|
"\"notifications/cancelled\"}}"));
|
|
ASSERT_FALSE(cbm_daemon_frontend_is_cancellation_notification("not-json"));
|
|
ASSERT_FALSE(cbm_daemon_frontend_is_cancellation_notification(NULL));
|
|
PASS();
|
|
}
|
|
|
|
#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS
|
|
/* An idle frontend has one independent maintenance observer. Hold that
|
|
* observer before its first secure presence read, then require the real worker
|
|
* to complete an idle cycle. Any worker-side presence read is therefore an
|
|
* exact duplicate, independent of elapsed time or scheduler cadence. */
|
|
TEST(daemon_frontend_idle_uses_one_maintenance_observer) {
|
|
frontend_idle_fixture_t fixture;
|
|
bool fixture_ready = frontend_idle_fixture_start(&fixture);
|
|
int input_pipe[2] = {-1, -1};
|
|
bool pipe_ready = fixture_ready && cbm_pipe(input_pipe) == 0;
|
|
FILE *input = pipe_ready ? frontend_test_fdopen_read(input_pipe[0]) : NULL;
|
|
FILE *output = input ? tmpfile() : NULL;
|
|
cbm_daemon_frontend_session_config_t session = {
|
|
.bootstrap = {.role = CBM_DAEMON_PROCESS_MCP_CLIENT,
|
|
.endpoint = fixture.maintenance.endpoint,
|
|
.identity = &fixture.identity,
|
|
.executable_path = "unused",
|
|
.connect_timeout_ms = 30000,
|
|
.startup_timeout_ms = 30000},
|
|
.session_root = fixture.maintenance.parent,
|
|
.tool_profile = CBM_MCP_TOOL_PROFILE_ALL,
|
|
};
|
|
frontend_idle_run_t run = {
|
|
.client = fixture.client,
|
|
.manager = fixture.maintenance.manager,
|
|
.session = &session,
|
|
.input = input,
|
|
.output = output,
|
|
.result = -1,
|
|
};
|
|
if (output) {
|
|
fixture.client = NULL; /* cbm_daemon_frontend_mcp_run consumes it. */
|
|
}
|
|
cbm_daemon_frontend_test_observer_reset(true);
|
|
cbm_thread_t frontend_thread;
|
|
bool thread_started =
|
|
output && cbm_thread_create(&frontend_thread, 0, frontend_idle_run, &run) == 0;
|
|
uint64_t deadline = cbm_now_ms() + 10000U;
|
|
while (thread_started && cbm_now_ms() < deadline &&
|
|
(!cbm_daemon_frontend_test_monitor_waiting() ||
|
|
cbm_daemon_frontend_test_worker_idle_cycles() == 0)) {
|
|
cbm_usleep(1000);
|
|
}
|
|
bool monitor_waiting = cbm_daemon_frontend_test_monitor_waiting();
|
|
uint64_t monitor_observations = cbm_daemon_frontend_test_monitor_observations();
|
|
uint64_t worker_observations = cbm_daemon_frontend_test_worker_observations();
|
|
uint64_t idle_cycles = cbm_daemon_frontend_test_worker_idle_cycles();
|
|
cbm_daemon_frontend_test_observer_release();
|
|
bool input_closed = !pipe_ready || frontend_test_close_fd(input_pipe[1]) == 0;
|
|
bool joined = thread_started && cbm_thread_join(&frontend_thread) == 0;
|
|
bool input_stream_closed = !input || fclose(input) == 0;
|
|
bool output_closed = !output || fclose(output) == 0;
|
|
bool fixture_closed = frontend_idle_fixture_finish(&fixture);
|
|
|
|
ASSERT_TRUE(fixture_ready);
|
|
ASSERT_TRUE(pipe_ready);
|
|
ASSERT_TRUE(thread_started);
|
|
ASSERT_TRUE(monitor_waiting);
|
|
ASSERT_EQ(monitor_observations, 1);
|
|
ASSERT_TRUE(idle_cycles > 0);
|
|
ASSERT_EQ(worker_observations, 0);
|
|
ASSERT_TRUE(input_closed);
|
|
ASSERT_TRUE(joined);
|
|
ASSERT_EQ(run.result, 0);
|
|
ASSERT_TRUE(input_stream_closed);
|
|
ASSERT_TRUE(output_closed);
|
|
ASSERT_TRUE(fixture_closed);
|
|
PASS();
|
|
}
|
|
#endif
|
|
|
|
#ifndef _WIN32
|
|
/* RED: an MCP frontend's main thread can remain blocked forever in stdio while
|
|
* install/update/uninstall owns maintenance intent. The already-present
|
|
* frontend worker must observe that native intent and terminate this stateless
|
|
* child, releasing its cohort lease and kernel IPC ownership. A deliberately
|
|
* invalid client pointer is safe only if the maintenance path exits before
|
|
* ordinary EOF/session-close handling, which also pins that exact path. */
|
|
TEST(daemon_frontend_maintenance_exits_while_stdio_reader_is_blocked) {
|
|
frontend_maintenance_fixture_t fixture;
|
|
ASSERT_TRUE(frontend_maintenance_fixture_start(&fixture, "blocked-stdio"));
|
|
int input_pipe[2] = {-1, -1};
|
|
int ready_pipe[2] = {-1, -1};
|
|
bool pipes_ready = pipe(input_pipe) == 0 && pipe(ready_pipe) == 0;
|
|
pid_t child = pipes_ready ? fork() : -1;
|
|
if (child == 0) {
|
|
(void)close(input_pipe[1]);
|
|
(void)close(ready_pipe[0]);
|
|
cbm_daemon_ipc_endpoint_t *endpoint =
|
|
cbm_daemon_ipc_endpoint_new("0123456789abcdef", fixture.parent);
|
|
cbm_version_cohort_manager_t *manager =
|
|
endpoint ? cbm_version_cohort_manager_new(endpoint) : NULL;
|
|
cbm_version_cohort_lease_t *participant = NULL;
|
|
cbm_daemon_conflict_t conflict;
|
|
cbm_daemon_build_identity_t identity = frontend_test_identity();
|
|
cbm_version_cohort_status_t admitted =
|
|
manager ? cbm_version_cohort_acquire(manager, &identity, cbm_now_ms() + 2000U,
|
|
&participant, &conflict)
|
|
: CBM_VERSION_COHORT_IO;
|
|
FILE *input = admitted == CBM_VERSION_COHORT_OK ? fdopen(input_pipe[0], "rb") : NULL;
|
|
FILE *output = input ? tmpfile() : NULL;
|
|
const char ready = 'R';
|
|
bool announced = output && write(ready_pipe[1], &ready, 1) == 1;
|
|
(void)close(ready_pipe[1]);
|
|
if (!announced) {
|
|
_exit(70);
|
|
}
|
|
cbm_daemon_frontend_session_config_t session = {
|
|
.bootstrap =
|
|
{
|
|
.role = CBM_DAEMON_PROCESS_MCP_CLIENT,
|
|
.endpoint = endpoint,
|
|
.identity = &identity,
|
|
.executable_path = "unused",
|
|
.connect_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS,
|
|
.startup_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS,
|
|
},
|
|
.session_root = fixture.parent,
|
|
.tool_profile = CBM_MCP_TOOL_PROFILE_ALL,
|
|
};
|
|
int result = cbm_daemon_frontend_mcp_run((cbm_daemon_runtime_client_t *)(uintptr_t)1,
|
|
manager, &session, input, output);
|
|
(void)result;
|
|
_exit(71);
|
|
}
|
|
|
|
if (pipes_ready) {
|
|
(void)close(input_pipe[0]);
|
|
(void)close(ready_pipe[1]);
|
|
}
|
|
bool announced = child > 0 && frontend_test_wait_byte(ready_pipe[0], 'R', cbm_now_ms() + 2000U);
|
|
cbm_version_cohort_lease_t *mutation = NULL;
|
|
cbm_version_cohort_quiesce_result_t quiesce = CBM_VERSION_COHORT_QUIESCE_NOT_NEEDED;
|
|
cbm_version_cohort_status_t status =
|
|
announced ? cbm_version_cohort_reserve_for_mutation(fixture.manager, cbm_now_ms() + 3000U,
|
|
frontend_test_quiesce_requested, NULL,
|
|
&quiesce, &mutation)
|
|
: CBM_VERSION_COHORT_IO;
|
|
if (status != CBM_VERSION_COHORT_OK && child > 0) {
|
|
(void)kill(child, SIGKILL);
|
|
}
|
|
int child_status = 0;
|
|
bool waited = child > 0 && waitpid(child, &child_status, 0) == child;
|
|
if (pipes_ready) {
|
|
(void)close(input_pipe[1]);
|
|
(void)close(ready_pipe[0]);
|
|
}
|
|
frontend_test_release_lease(&mutation);
|
|
frontend_maintenance_fixture_finish(&fixture);
|
|
|
|
ASSERT_TRUE(pipes_ready);
|
|
ASSERT_TRUE(child > 0);
|
|
ASSERT_TRUE(announced);
|
|
ASSERT_EQ(status, CBM_VERSION_COHORT_OK);
|
|
ASSERT_EQ(quiesce, CBM_VERSION_COHORT_QUIESCE_REQUESTED);
|
|
ASSERT_TRUE(waited);
|
|
ASSERT_TRUE(WIFEXITED(child_status));
|
|
ASSERT_EQ(WEXITSTATUS(child_status), 0);
|
|
PASS();
|
|
}
|
|
|
|
/* RED: local CLI/worker work has no standing daemon session to interrupt. Its
|
|
* command-lifetime monitor must request cooperative cancellation once, allow a
|
|
* bounded grace, then hard-exit the isolated process so SQLite/native locks
|
|
* roll back and the mutation barrier can prove that every lifetime participant
|
|
* is gone. */
|
|
TEST(daemon_local_participant_monitor_cancels_then_bounds_active_operation) {
|
|
frontend_maintenance_fixture_t fixture;
|
|
ASSERT_TRUE(frontend_maintenance_fixture_start(&fixture, "local-monitor"));
|
|
int ready_pipe[2] = {-1, -1};
|
|
int cancel_pipe[2] = {-1, -1};
|
|
bool pipes_ready = pipe(ready_pipe) == 0 && pipe(cancel_pipe) == 0;
|
|
pid_t child = pipes_ready ? fork() : -1;
|
|
if (child == 0) {
|
|
(void)close(ready_pipe[0]);
|
|
(void)close(cancel_pipe[0]);
|
|
cbm_daemon_ipc_endpoint_t *endpoint =
|
|
cbm_daemon_ipc_endpoint_new("0123456789abcdef", fixture.parent);
|
|
cbm_version_cohort_manager_t *manager =
|
|
endpoint ? cbm_version_cohort_manager_new(endpoint) : NULL;
|
|
cbm_version_cohort_lease_t *participant = NULL;
|
|
cbm_daemon_conflict_t conflict;
|
|
cbm_daemon_build_identity_t identity = frontend_test_identity();
|
|
cbm_version_cohort_status_t admitted =
|
|
manager ? cbm_version_cohort_acquire(manager, &identity, cbm_now_ms() + 2000U,
|
|
&participant, &conflict)
|
|
: CBM_VERSION_COHORT_IO;
|
|
cbm_daemon_maintenance_monitor_t *monitor =
|
|
admitted == CBM_VERSION_COHORT_OK
|
|
? cbm_daemon_maintenance_monitor_start(manager, frontend_test_cancel_active,
|
|
&cancel_pipe[1], 37, "test-local-operation")
|
|
: NULL;
|
|
const char ready = 'R';
|
|
bool announced = monitor && write(ready_pipe[1], &ready, 1) == 1;
|
|
(void)close(ready_pipe[1]);
|
|
if (!announced) {
|
|
_exit(72);
|
|
}
|
|
for (;;) {
|
|
cbm_usleep(100000);
|
|
}
|
|
}
|
|
|
|
if (pipes_ready) {
|
|
(void)close(ready_pipe[1]);
|
|
(void)close(cancel_pipe[1]);
|
|
}
|
|
bool announced = child > 0 && frontend_test_wait_byte(ready_pipe[0], 'R', cbm_now_ms() + 2000U);
|
|
cbm_version_cohort_lease_t *mutation = NULL;
|
|
cbm_version_cohort_quiesce_result_t quiesce = CBM_VERSION_COHORT_QUIESCE_NOT_NEEDED;
|
|
cbm_version_cohort_status_t status =
|
|
announced ? cbm_version_cohort_reserve_for_mutation(fixture.manager, cbm_now_ms() + 5000U,
|
|
frontend_test_quiesce_requested, NULL,
|
|
&quiesce, &mutation)
|
|
: CBM_VERSION_COHORT_IO;
|
|
bool cancelled = status == CBM_VERSION_COHORT_OK &&
|
|
frontend_test_wait_byte(cancel_pipe[0], 'C', cbm_now_ms() + 1000U);
|
|
if (status != CBM_VERSION_COHORT_OK && child > 0) {
|
|
(void)kill(child, SIGKILL);
|
|
}
|
|
int child_status = 0;
|
|
bool waited = child > 0 && waitpid(child, &child_status, 0) == child;
|
|
if (pipes_ready) {
|
|
(void)close(ready_pipe[0]);
|
|
(void)close(cancel_pipe[0]);
|
|
}
|
|
frontend_test_release_lease(&mutation);
|
|
frontend_maintenance_fixture_finish(&fixture);
|
|
|
|
ASSERT_TRUE(pipes_ready);
|
|
ASSERT_TRUE(child > 0);
|
|
ASSERT_TRUE(announced);
|
|
ASSERT_EQ(status, CBM_VERSION_COHORT_OK);
|
|
ASSERT_EQ(quiesce, CBM_VERSION_COHORT_QUIESCE_REQUESTED);
|
|
ASSERT_TRUE(cancelled);
|
|
ASSERT_TRUE(waited);
|
|
ASSERT_TRUE(WIFEXITED(child_status));
|
|
ASSERT_EQ(WEXITSTATUS(child_status), 37);
|
|
PASS();
|
|
}
|
|
|
|
/* RED: supervised process-tree cancellation permits one second of graceful
|
|
* shutdown followed by one second of forced containment. The maintenance
|
|
* observer must not _Exit the owning process before that bounded supervisor
|
|
* window can finish and join the observer normally. */
|
|
TEST(daemon_local_participant_monitor_allows_supervisor_containment_window) {
|
|
frontend_maintenance_fixture_t fixture;
|
|
ASSERT_TRUE(frontend_maintenance_fixture_start(&fixture, "monitor-window"));
|
|
int ready_pipe[2] = {-1, -1};
|
|
bool pipe_ready = pipe(ready_pipe) == 0;
|
|
pid_t child = pipe_ready ? fork() : -1;
|
|
if (child == 0) {
|
|
(void)close(ready_pipe[0]);
|
|
cbm_daemon_ipc_endpoint_t *endpoint =
|
|
cbm_daemon_ipc_endpoint_new("0123456789abcdef", fixture.parent);
|
|
cbm_version_cohort_manager_t *manager =
|
|
endpoint ? cbm_version_cohort_manager_new(endpoint) : NULL;
|
|
cbm_version_cohort_lease_t *participant = NULL;
|
|
cbm_daemon_conflict_t conflict;
|
|
cbm_daemon_build_identity_t identity = frontend_test_identity();
|
|
cbm_version_cohort_status_t admitted =
|
|
manager ? cbm_version_cohort_acquire(manager, &identity, cbm_now_ms() + 2000U,
|
|
&participant, &conflict)
|
|
: CBM_VERSION_COHORT_IO;
|
|
atomic_bool cancelled;
|
|
atomic_init(&cancelled, false);
|
|
cbm_daemon_maintenance_monitor_t *monitor =
|
|
admitted == CBM_VERSION_COHORT_OK
|
|
? cbm_daemon_maintenance_monitor_start(manager, frontend_test_cancel_and_mark,
|
|
&cancelled, 38, "test-supervisor-window")
|
|
: NULL;
|
|
const char ready = 'R';
|
|
bool announced = monitor && write(ready_pipe[1], &ready, 1) == 1;
|
|
(void)close(ready_pipe[1]);
|
|
if (!announced) {
|
|
_exit(72);
|
|
}
|
|
while (!atomic_load_explicit(&cancelled, memory_order_acquire)) {
|
|
cbm_usleep(1000);
|
|
}
|
|
|
|
/* Model the maximum graceful + forced-settle supervisor bounds. */
|
|
cbm_usleep(2100U * 1000U);
|
|
bool stopped = cbm_daemon_maintenance_monitor_stop(&monitor);
|
|
frontend_test_release_lease(&participant);
|
|
while (manager && cbm_version_cohort_manager_free(&manager) != CBM_PRIVATE_FILE_LOCK_OK) {
|
|
cbm_usleep(1000);
|
|
}
|
|
cbm_daemon_ipc_endpoint_free(endpoint);
|
|
_exit(stopped ? 0 : 73);
|
|
}
|
|
|
|
if (pipe_ready) {
|
|
(void)close(ready_pipe[1]);
|
|
}
|
|
bool announced = child > 0 && frontend_test_wait_byte(ready_pipe[0], 'R', cbm_now_ms() + 2000U);
|
|
cbm_version_cohort_lease_t *mutation = NULL;
|
|
cbm_version_cohort_quiesce_result_t quiesce = CBM_VERSION_COHORT_QUIESCE_NOT_NEEDED;
|
|
cbm_version_cohort_status_t status =
|
|
announced ? cbm_version_cohort_reserve_for_mutation(fixture.manager, cbm_now_ms() + 5000U,
|
|
frontend_test_quiesce_requested, NULL,
|
|
&quiesce, &mutation)
|
|
: CBM_VERSION_COHORT_IO;
|
|
if (status != CBM_VERSION_COHORT_OK && child > 0) {
|
|
(void)kill(child, SIGKILL);
|
|
}
|
|
int child_status = 0;
|
|
bool waited = child > 0 && waitpid(child, &child_status, 0) == child;
|
|
if (pipe_ready) {
|
|
(void)close(ready_pipe[0]);
|
|
}
|
|
frontend_test_release_lease(&mutation);
|
|
frontend_maintenance_fixture_finish(&fixture);
|
|
|
|
ASSERT_TRUE(pipe_ready);
|
|
ASSERT_TRUE(child > 0);
|
|
ASSERT_TRUE(announced);
|
|
ASSERT_EQ(status, CBM_VERSION_COHORT_OK);
|
|
ASSERT_EQ(quiesce, CBM_VERSION_COHORT_QUIESCE_REQUESTED);
|
|
ASSERT_TRUE(waited);
|
|
ASSERT_TRUE(WIFEXITED(child_status));
|
|
ASSERT_EQ(WEXITSTATUS(child_status), 0);
|
|
PASS();
|
|
}
|
|
|
|
/* RED: once a long request is active, a buffered writer can fill the bounded
|
|
* frontend queue and close its pipe. Waiting indefinitely for queue capacity
|
|
* prevents the sole reader from observing that already-pending EOF, so neither
|
|
* the request nor its daemon session is cancelled. Overload must instead fail
|
|
* the frontend promptly and close/cancel the authenticated session. */
|
|
TEST(daemon_frontend_over_capacity_input_backpressures_without_loss) {
|
|
ASSERT_TRUE(frontend_eof_run_isolated("overflow", true));
|
|
PASS();
|
|
}
|
|
|
|
/* A request rejected before its application frame reaches the daemon is safe
|
|
* to replay. The frontend must replace its stale authenticated client, restore
|
|
* the session, reserve a token from the new connection, and return the one
|
|
* response without closing Codex's stdio transport. */
|
|
TEST(daemon_frontend_recovers_one_unsent_request_after_daemon_replacement) {
|
|
ASSERT_TRUE(frontend_recovery_run_isolated(FRONTEND_RECOVERY_RETRY));
|
|
PASS();
|
|
}
|
|
|
|
/* Context-replay cancellation must not strand a token or break the next request. */
|
|
#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS
|
|
TEST(daemon_frontend_recovery_cancellation_preserves_next_request) {
|
|
ASSERT_TRUE(frontend_recovery_run_isolated(FRONTEND_RECOVERY_CANCEL));
|
|
PASS();
|
|
}
|
|
#endif
|
|
|
|
/* Clean EOF during context replay must preserve worker-owned client cleanup. */
|
|
TEST(daemon_frontend_clean_eof_during_recovery_preserves_client_ownership) {
|
|
ASSERT_TRUE(frontend_recovery_run_isolated(FRONTEND_RECOVERY_CLEAN_EOF));
|
|
PASS();
|
|
}
|
|
|
|
/* Clean EOF is the normal MCP-session ownership boundary. Accepted work gets a
|
|
* bounded drain opportunity; work still active at the deadline is cancelled
|
|
* with that exact session, while the frontend itself reports a clean close. */
|
|
TEST(daemon_frontend_eof_drain_timeout_cancels_and_returns_success) {
|
|
ASSERT_TRUE(frontend_eof_run_isolated("timeout", false));
|
|
PASS();
|
|
}
|
|
|
|
/* A daemon response can finish its IPC exchange and then block forever while
|
|
* writing to an agent that stopped reading stdout. EOF must still bound the
|
|
* thin frontend process. Keep the daemon in a separate child so the frontend's
|
|
* terminal watchdog closes a real kernel connection; the daemon must observe
|
|
* that close, cancel the exact session, and shut down normally. */
|
|
TEST(daemon_frontend_stdout_backpressure_eof_fail_stops_and_cancels_session) {
|
|
ASSERT_TRUE(frontend_backpressure_run_isolated(false));
|
|
PASS();
|
|
}
|
|
|
|
/* Maintenance must not depend on either MCP stdio thread making progress.
|
|
* Keep stdin open, fill stdout with a completed daemon response, then publish
|
|
* activation intent. The independent monitor must release the cohort lease and
|
|
* kernel session within its bound; the daemon must cancel that exact session. */
|
|
TEST(daemon_frontend_stdout_backpressure_maintenance_stops_and_cancels_session) {
|
|
ASSERT_TRUE(frontend_backpressure_run_isolated(true));
|
|
PASS();
|
|
}
|
|
#endif
|
|
|
|
TEST(daemon_local_participant_monitor_joins_before_manager_teardown) {
|
|
frontend_maintenance_fixture_t fixture;
|
|
ASSERT_TRUE(frontend_maintenance_fixture_start(&fixture, "monitor-join"));
|
|
cbm_daemon_maintenance_monitor_t *monitor = cbm_daemon_maintenance_monitor_start(
|
|
fixture.manager, NULL, NULL, EXIT_FAILURE, "test-idle-command");
|
|
bool started = monitor != NULL;
|
|
bool stopped = started && cbm_daemon_maintenance_monitor_stop(&monitor);
|
|
bool consumed = monitor == NULL;
|
|
frontend_maintenance_fixture_finish(&fixture);
|
|
|
|
ASSERT_TRUE(started);
|
|
ASSERT_TRUE(stopped);
|
|
ASSERT_TRUE(consumed);
|
|
PASS();
|
|
}
|
|
|
|
SUITE(daemon_frontend) {
|
|
RUN_TEST(daemon_frontend_recognizes_exact_cancellation_notification);
|
|
RUN_TEST(daemon_frontend_correlates_cancellation_to_exact_request);
|
|
RUN_TEST(daemon_frontend_cancelled_index_reply_offers_async_issue2144);
|
|
RUN_TEST(daemon_frontend_ignores_cancellation_text_in_string_content);
|
|
RUN_TEST(daemon_frontend_rejects_non_notification_cancellation_shapes);
|
|
#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS
|
|
RUN_TEST(daemon_frontend_idle_uses_one_maintenance_observer);
|
|
#endif
|
|
#ifndef _WIN32
|
|
RUN_TEST(daemon_frontend_maintenance_exits_while_stdio_reader_is_blocked);
|
|
RUN_TEST(daemon_local_participant_monitor_cancels_then_bounds_active_operation);
|
|
RUN_TEST(daemon_local_participant_monitor_allows_supervisor_containment_window);
|
|
RUN_TEST(daemon_frontend_over_capacity_input_backpressures_without_loss);
|
|
RUN_TEST(daemon_frontend_recovers_one_unsent_request_after_daemon_replacement);
|
|
#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS
|
|
RUN_TEST(daemon_frontend_recovery_cancellation_preserves_next_request);
|
|
#endif
|
|
RUN_TEST(daemon_frontend_clean_eof_during_recovery_preserves_client_ownership);
|
|
RUN_TEST(daemon_frontend_eof_drain_timeout_cancels_and_returns_success);
|
|
RUN_TEST(daemon_frontend_stdout_backpressure_eof_fail_stops_and_cancels_session);
|
|
RUN_TEST(daemon_frontend_stdout_backpressure_maintenance_stops_and_cancels_session);
|
|
#endif
|
|
RUN_TEST(daemon_local_participant_monitor_joins_before_manager_teardown);
|
|
}
|