Merge pull request #2488 from DeusData/test/daemon-runtime-connection-cap-deterministic

fix(daemon): preserve capacity rejection before the client HELLO
This commit is contained in:
Martin Vogel
2026-10-04 20:27:41 +02:00
committed by GitHub
5 changed files with 272 additions and 39 deletions
+18
View File
@@ -7078,3 +7078,21 @@ int cbm_daemon_ipc_receive_frame(cbm_daemon_ipc_connection_t *connection, uint32
return cbm_daemon_ipc_receive_frame_bounded(connection, timeout_ms, CBM_DAEMON_MAX_FRAME_SIZE,
frame_out, payload_out);
}
int cbm_daemon_ipc_receive_frame_after_failed_send(cbm_daemon_ipc_connection_t *connection,
uint32_t timeout_ms, uint32_t max_payload_length,
cbm_daemon_frame_t *frame_out,
uint8_t **payload_out) {
if (!connection) {
return cbm_daemon_ipc_receive_frame_bounded(NULL, timeout_ms, max_payload_length, frame_out,
payload_out);
}
/* The poison came from the failed send alone (nothing was read yet), and it
* guards the OUTBOUND frame boundary; the inbound one is intact. Lift it for
* this one receive, then restore it so no later send can reuse the stream. */
atomic_store_explicit(&connection->poisoned, false, memory_order_release);
int received = cbm_daemon_ipc_receive_frame_bounded(connection, timeout_ms, max_payload_length,
frame_out, payload_out);
atomic_store_explicit(&connection->poisoned, true, memory_order_release);
return received;
}
+16
View File
@@ -75,6 +75,22 @@ int cbm_daemon_ipc_receive_frame_bounded(cbm_daemon_ipc_connection_t *connection
uint32_t timeout_ms, uint32_t max_payload_length,
cbm_daemon_frame_t *frame_out, uint8_t **payload_out);
/* Bounded receive of an answer the peer queued before it closed, on a
* connection whose send just failed. A daemon at capacity or stopping answers
* from its accept loop without reading the request and closes at once; when
* that close lands before the client's first send, the send fails (EPIPE) and
* poisons the connection although the complete answer is still queued for
* reading -- POSIX keeps it readable after the peer's close (a Windows server
* close discards it, so there this fails like any other receive). A failed
* send consumes no inbound bytes, so the inbound frame boundary is intact;
* call this ONLY before anything was read from the connection. The connection
* stays poisoned for sends afterwards. Returns like
* cbm_daemon_ipc_receive_frame_bounded. */
int cbm_daemon_ipc_receive_frame_after_failed_send(cbm_daemon_ipc_connection_t *connection,
uint32_t timeout_ms, uint32_t max_payload_length,
cbm_daemon_frame_t *frame_out,
uint8_t **payload_out);
/* Narrow crash/fault seams for publication-state and retry-state tests. They
* are inert unless a test installs a hook/failure count in its own process. */
typedef enum {
+72 -8
View File
@@ -76,8 +76,26 @@ static _Atomic uint32_t runtime_ephemeral_linger_timeout_seam = UINT32_MAX;
void cbm_daemon_runtime_service_set_ephemeral_linger_timeout_for_testing(uint32_t timeout_ms) {
atomic_store(&runtime_ephemeral_linger_timeout_seam, timeout_ms);
}
/* Connection-cap seams (flaky-ledger item 12): see runtime.h. */
static atomic_bool runtime_hold_peer_waits_seam;
void cbm_daemon_runtime_hold_peer_waits_for_testing(bool hold) {
atomic_store(&runtime_hold_peer_waits_seam, hold);
}
static _Atomic(cbm_daemon_runtime_hello_hook_t) runtime_hello_hook_seam;
void cbm_daemon_runtime_set_hello_hook_for_testing(cbm_daemon_runtime_hello_hook_t hook) {
atomic_store(&runtime_hello_hook_seam, hook);
}
static void runtime_hello_point(cbm_daemon_runtime_hello_point_t point) {
cbm_daemon_runtime_hello_hook_t hook = atomic_load(&runtime_hello_hook_seam);
if (hook) {
hook(point);
}
}
#define RUNTIME_HELLO_POINT(point) runtime_hello_point(CBM_DAEMON_RUNTIME_HELLO_POINT_##point)
#else
static void runtime_note_peer_image_hash(void) {}
#define RUNTIME_HELLO_POINT(point) ((void)0)
#endif
#ifdef _WIN32
@@ -1989,6 +2007,29 @@ static void runtime_worker_handle_activation_shutdown(cbm_daemon_runtime_worker_
runtime_worker_finish(worker);
}
/* The first frame is due within the request deadline. Under the item-12 seam a
* connection that never sends one holds its slot until it closes; stop still
* interrupts the wait like any other worker receive. */
static uint32_t runtime_first_frame_timeout_ms(const cbm_daemon_runtime_service_t *service) {
#ifdef CBM_ENABLE_TEST_SEAMS
if (atomic_load(&runtime_hold_peer_waits_seam)) {
return CBM_DAEMON_IPC_WAIT_FOREVER;
}
#endif
return service->request_timeout_ms;
}
/* Under the same seam an inline rejection's drain ends only on the peer's
* close, never on the 250 ms bound (the drain only waits on Windows). */
static uint32_t runtime_reject_drain_timeout_ms(void) {
#ifdef CBM_ENABLE_TEST_SEAMS
if (atomic_load(&runtime_hold_peer_waits_seam)) {
return CBM_DAEMON_IPC_WAIT_FOREVER;
}
#endif
return RUNTIME_REJECT_DRAIN_TIMEOUT_MS;
}
static void *runtime_connection_worker(void *opaque) {
cbm_daemon_runtime_worker_t *worker = opaque;
cbm_daemon_runtime_service_t *service = worker->service;
@@ -1998,7 +2039,7 @@ static void *runtime_connection_worker(void *opaque) {
char requested_build[CBM_DAEMON_BUILD_FINGERPRINT_SIZE];
int received = cbm_daemon_ipc_receive_frame_bounded(
worker->connection, service->request_timeout_ms,
worker->connection, runtime_first_frame_timeout_ms(service),
CBM_DAEMON_ACTIVATION_SHUTDOWN_REQUEST_SIZE, &frame, &payload);
if (received != 1 || frame.type != CBM_DAEMON_FRAME_REQUEST) {
free(payload);
@@ -2274,14 +2315,19 @@ static cbm_daemon_runtime_worker_t *runtime_find_free_worker_locked(
return NULL;
}
/* Answers WITHOUT reading the peer's HELLO, so on POSIX (where the drain is a
* no-op) the close can land before the peer has even sent it; the client reads
* this answer after its failed send (cbm_daemon_runtime_client_connect). */
static void runtime_reject_inline(cbm_daemon_ipc_connection_t *connection, const char *message) {
cbm_daemon_runtime_connect_result_t result;
runtime_result_rejected(&result, message);
RUNTIME_HELLO_POINT(DAEMON_REJECTING);
(void)runtime_send_hello_response(connection, &result);
/* Same Windows named-pipe discard hazard as runtime_worker_finish: the
* peer must get to read the rejection before the handle closes. */
cbm_daemon_ipc_connection_drain(connection, RUNTIME_REJECT_DRAIN_TIMEOUT_MS);
cbm_daemon_ipc_connection_drain(connection, runtime_reject_drain_timeout_ms());
cbm_daemon_ipc_connection_close(connection);
RUNTIME_HELLO_POINT(DAEMON_REJECTED);
}
static void runtime_accept_connection(cbm_daemon_runtime_service_t *service,
@@ -3063,21 +3109,39 @@ cbm_daemon_runtime_client_t *cbm_daemon_runtime_client_connect(
return NULL;
}
cbm_daemon_ipc_connection_t *connection = cbm_daemon_ipc_connect(endpoint, timeout_ms);
if (!connection || !cbm_daemon_ipc_send_frame(connection, CBM_DAEMON_FRAME_REQUEST,
CBM_DAEMON_RUNTIME_OP_HELLO, request,
(uint32_t)sizeof(request))) {
cbm_daemon_ipc_connection_close(connection);
if (!connection) {
return NULL;
}
RUNTIME_HELLO_POINT(CLIENT_CONNECTED);
bool hello_sent =
cbm_daemon_ipc_send_frame(connection, CBM_DAEMON_FRAME_REQUEST, CBM_DAEMON_RUNTIME_OP_HELLO,
request, (uint32_t)sizeof(request));
RUNTIME_HELLO_POINT(CLIENT_SENT);
cbm_daemon_frame_t frame = {0};
uint8_t *payload = NULL;
int received = cbm_daemon_ipc_receive_frame_bounded(
connection, timeout_ms, CBM_DAEMON_RENDEZVOUS_RESPONSE_SIZE, &frame, &payload);
/* A daemon at capacity or stopping answers from its accept loop without
* reading this HELLO and closes at once. When that close lands first, the
* send fails (EPIPE) with the whole answer already queued here: read it,
* or the caller gets a bare transport error and bootstrap classifies by
* lock state instead of by the daemon's own "capacity" / "stopping". */
int received =
hello_sent
? cbm_daemon_ipc_receive_frame_bounded(
connection, timeout_ms, CBM_DAEMON_RENDEZVOUS_RESPONSE_SIZE, &frame, &payload)
: cbm_daemon_ipc_receive_frame_after_failed_send(
connection, timeout_ms, CBM_DAEMON_RENDEZVOUS_RESPONSE_SIZE, &frame, &payload);
bool valid = received == 1 && frame.type == CBM_DAEMON_FRAME_RESPONSE &&
frame.flags == CBM_DAEMON_RUNTIME_OP_HELLO &&
frame.length == CBM_DAEMON_RENDEZVOUS_RESPONSE_SIZE &&
runtime_hello_response_decode(payload, result_out);
free(payload);
if (!hello_sent && (!valid || result_out->status != CBM_DAEMON_RUNTIME_CONNECT_REJECTED)) {
/* Only a rejection can answer a HELLO the daemon never received;
* anything else stays the plain send failure it always was. */
memset(result_out, 0, sizeof(*result_out));
cbm_daemon_ipc_connection_close(connection);
return NULL;
}
if (received != 1) {
/* The kernel completed the pipe/socket connect, so a server process
* exists, yet the HELLO went unanswered. Name that holder: a dead
+21
View File
@@ -471,6 +471,27 @@ void cbm_daemon_runtime_set_containment_hook_for_testing(
* in test time. UINT32_MAX restores the production constant; any other value
* (0 = expire immediately) overrides. Process-global; reset it after use. */
void cbm_daemon_runtime_service_set_ephemeral_linger_timeout_for_testing(uint32_t timeout_ms);
/* Connection-cap seams (flaky-ledger item 12). While held, the two bounded
* waits on a peer in the connection-cap path end only on that peer: a
* connection that has not sent its first frame keeps its worker slot until it
* closes (or the service stops) instead of until the request deadline, and an
* inline rejection's drain (Windows; a no-op on POSIX) lasts until the peer
* closes instead of 250 ms. A "slow hello" becomes a state the test releases,
* not a window it races. Process-global; reset it after use. */
void cbm_daemon_runtime_hold_peer_waits_for_testing(bool hold);
/* The four points that order a HELLO against an inline accept-loop rejection.
* The hook runs on the client thread for the CLIENT_* points and on the accept
* thread for the DAEMON_* points (never under the service mutex), so a test can
* pin either order: daemon answers after the HELLO is on the wire, or the
* daemon has answered and closed before the HELLO is sent. NULL removes it. */
typedef enum {
CBM_DAEMON_RUNTIME_HELLO_POINT_CLIENT_CONNECTED = 0, /* transport up, HELLO not sent */
CBM_DAEMON_RUNTIME_HELLO_POINT_CLIENT_SENT, /* HELLO send attempted */
CBM_DAEMON_RUNTIME_HELLO_POINT_DAEMON_REJECTING, /* inline rejection not yet written */
CBM_DAEMON_RUNTIME_HELLO_POINT_DAEMON_REJECTED, /* inline rejection written + closed */
} cbm_daemon_runtime_hello_point_t;
typedef void (*cbm_daemon_runtime_hello_hook_t)(cbm_daemon_runtime_hello_point_t point);
void cbm_daemon_runtime_set_hello_hook_for_testing(cbm_daemon_runtime_hello_hook_t hook);
#endif
#endif /* CBM_DAEMON_RUNTIME_H */
+145 -31
View File
@@ -2827,60 +2827,138 @@ TEST(daemon_runtime_authenticated_idle_connection_outlives_lease_interval) {
PASS();
}
TEST(daemon_runtime_connection_cap_covers_slow_hello_and_stopping_is_terminal) {
/* Flaky-ledger item 12 (7 sightings, ASSERT(capacity_rejected)): the overflow
* client's HELLO raced the daemon's inline capacity rejection, which the
* accept loop writes WITHOUT reading that HELLO and follows with an immediate
* close. The hook pins one order per run so no window is left to race; the
* hold seam keeps the slow hello's slot until the test closes it, where the
* request deadline used to free it after 2 s. The waits below end on events
* that are guaranteed to arrive (see each comment), never on a bound. */
typedef enum {
RUNTIME_CAP_HELLO_BEFORE_REJECTION = 0,
RUNTIME_CAP_REJECTION_BEFORE_HELLO = 1,
} runtime_cap_order_t;
static atomic_int runtime_cap_order;
static atomic_bool runtime_cap_hello_sent;
static atomic_bool runtime_cap_rejection_closed;
static atomic_bool runtime_cap_order_pinned;
static cbm_daemon_runtime_service_t *runtime_cap_service;
static void runtime_cap_hello_hook(cbm_daemon_runtime_hello_point_t point) {
int order = atomic_load(&runtime_cap_order);
if (point == CBM_DAEMON_RUNTIME_HELLO_POINT_CLIENT_SENT) {
atomic_store(&runtime_cap_hello_sent, true);
} else if (point == CBM_DAEMON_RUNTIME_HELLO_POINT_DAEMON_REJECTED) {
atomic_store(&runtime_cap_rejection_closed, true);
} else if (point == CBM_DAEMON_RUNTIME_HELLO_POINT_DAEMON_REJECTING &&
order == RUNTIME_CAP_HELLO_BEFORE_REJECTION) {
/* Accept thread. The client always gets there: its connect already
* completed and a HELLO fits the transport buffer, so its send never
* waits on this thread. */
while (!atomic_load(&runtime_cap_hello_sent)) {
cbm_usleep(200);
}
atomic_store(&runtime_cap_order_pinned, true);
} else if (point == CBM_DAEMON_RUNTIME_HELLO_POINT_CLIENT_CONNECTED &&
order == RUNTIME_CAP_REJECTION_BEFORE_HELLO) {
/* Client (test) thread. Ends on the rejection, or on an admission (a
* broken cap): that worker is held by the seam, so a third slot stays
* counted until this HELLO goes out and the test fails on ACCEPTED. */
while (!atomic_load(&runtime_cap_rejection_closed) &&
cbm_daemon_runtime_service_active_connections(runtime_cap_service) <= 2) {
cbm_usleep(200);
}
atomic_store(&runtime_cap_order_pinned, atomic_load(&runtime_cap_rejection_closed));
}
}
typedef struct {
bool started;
bool slow_slot_counted;
bool order_pinned;
bool capacity_rejected;
bool slow_slot_released;
bool exited;
bool no_resurrection;
} runtime_cap_outcome_t;
static cbm_daemon_runtime_client_t *runtime_cap_overflow_connect(
runtime_test_fixture_t *fixture, const cbm_daemon_build_identity_t *identity,
runtime_cap_order_t order, runtime_cap_outcome_t *outcome) {
cbm_daemon_runtime_connect_result_t result = {0};
runtime_cap_service = fixture->service;
atomic_store(&runtime_cap_order, (int)order);
atomic_store(&runtime_cap_hello_sent, false);
atomic_store(&runtime_cap_rejection_closed, false);
atomic_store(&runtime_cap_order_pinned, false);
cbm_daemon_runtime_set_hello_hook_for_testing(runtime_cap_hello_hook);
cbm_daemon_runtime_client_t *overflow = cbm_daemon_runtime_client_connect(
fixture->endpoint, identity, RUNTIME_TEST_TIMEOUT_MS, &result);
cbm_daemon_runtime_set_hello_hook_for_testing(NULL);
/* Both counts are HELD here: the slow hello's slot by the seam, the
* accepted client by its open session. */
size_t connections = cbm_daemon_runtime_service_active_connections(fixture->service);
size_t clients = cbm_daemon_runtime_service_active_clients(fixture->service);
outcome->order_pinned = atomic_load(&runtime_cap_order_pinned);
outcome->capacity_rejected =
overflow == NULL && result.status == CBM_DAEMON_RUNTIME_CONNECT_REJECTED &&
strstr(result.message, "capacity") != NULL && connections == 2 && clients == 1;
if (!outcome->capacity_rejected) {
printf(" overflow: client=%s status=%d message=\"%s\" active_connections=%zu "
"active_clients=%zu order=%d pinned=%d\n",
overflow ? "admitted" : "null", (int)result.status, result.message, connections,
clients, (int)order, outcome->order_pinned);
}
return overflow;
}
static void runtime_cap_scenario(runtime_cap_order_t order, runtime_cap_outcome_t *outcome) {
memset(outcome, 0, sizeof(*outcome));
cbm_daemon_build_identity_t identity =
runtime_test_identity("2.4.0", runtime_test_self_build());
runtime_test_fixture_t fixture;
bool started = runtime_test_fixture_start_limited(&fixture, "connection-cap", &identity, 2);
cbm_daemon_runtime_hold_peer_waits_for_testing(true);
outcome->started = runtime_test_fixture_start_limited(&fixture, "connection-cap", &identity, 2);
cbm_daemon_ipc_connection_t *slow_hello = NULL;
cbm_daemon_runtime_client_t *accepted = NULL;
cbm_daemon_runtime_client_t *overflow = NULL;
cbm_daemon_runtime_client_t *resurrection = NULL;
cbm_daemon_runtime_connect_result_t accepted_result = {0};
cbm_daemon_runtime_connect_result_t overflow_result = {0};
cbm_daemon_runtime_connect_result_t resurrection_result = {0};
bool slow_slot_counted = false;
bool capacity_rejected = false;
bool slow_slot_released = false;
bool exited = false;
bool no_resurrection = false;
if (started) {
if (outcome->started) {
slow_hello = cbm_daemon_ipc_connect(fixture.endpoint, RUNTIME_TEST_TIMEOUT_MS);
}
if (slow_hello) {
slow_slot_counted = cbm_daemon_runtime_service_wait_for_connections(
outcome->slow_slot_counted = cbm_daemon_runtime_service_wait_for_connections(
fixture.service, 1, RUNTIME_TEST_TIMEOUT_MS);
accepted = cbm_daemon_runtime_client_connect(fixture.endpoint, &identity,
RUNTIME_TEST_TIMEOUT_MS, &accepted_result);
}
if (accepted) {
overflow = cbm_daemon_runtime_client_connect(fixture.endpoint, &identity,
RUNTIME_TEST_TIMEOUT_MS, &overflow_result);
capacity_rejected = overflow == NULL &&
overflow_result.status == CBM_DAEMON_RUNTIME_CONNECT_REJECTED &&
strstr(overflow_result.message, "capacity") != NULL &&
cbm_daemon_runtime_service_active_connections(fixture.service) == 2 &&
cbm_daemon_runtime_service_active_clients(fixture.service) == 1;
overflow = runtime_cap_overflow_connect(&fixture, &identity, order, outcome);
cbm_daemon_ipc_connection_close(slow_hello);
slow_hello = NULL;
slow_slot_released = cbm_daemon_runtime_service_wait_for_connections(
outcome->slow_slot_released = cbm_daemon_runtime_service_wait_for_connections(
fixture.service, 1, RUNTIME_TEST_TIMEOUT_MS);
(void)cbm_daemon_runtime_client_close(accepted, RUNTIME_TEST_TIMEOUT_MS);
accepted = NULL;
exited = cbm_daemon_runtime_service_wait_exited(fixture.service, RUNTIME_TEST_TIMEOUT_MS);
outcome->exited =
cbm_daemon_runtime_service_wait_exited(fixture.service, RUNTIME_TEST_TIMEOUT_MS);
resurrection = cbm_daemon_runtime_client_connect(fixture.endpoint, &identity, 100,
&resurrection_result);
cbm_daemon_runtime_service_state_t terminal_state =
cbm_daemon_runtime_service_state(fixture.service);
no_resurrection = resurrection == NULL &&
(resurrection_result.status == CBM_DAEMON_RUNTIME_CONNECT_ERROR ||
resurrection_result.status == CBM_DAEMON_RUNTIME_CONNECT_REJECTED) &&
cbm_daemon_runtime_service_active_clients(fixture.service) == 0 &&
(terminal_state == CBM_DAEMON_RUNTIME_SERVICE_STOPPING ||
terminal_state == CBM_DAEMON_RUNTIME_SERVICE_EXITED);
outcome->no_resurrection =
resurrection == NULL &&
(resurrection_result.status == CBM_DAEMON_RUNTIME_CONNECT_ERROR ||
resurrection_result.status == CBM_DAEMON_RUNTIME_CONNECT_REJECTED) &&
cbm_daemon_runtime_service_active_clients(fixture.service) == 0 &&
(terminal_state == CBM_DAEMON_RUNTIME_SERVICE_STOPPING ||
terminal_state == CBM_DAEMON_RUNTIME_SERVICE_EXITED);
}
if (resurrection) {
@@ -2894,16 +2972,49 @@ TEST(daemon_runtime_connection_cap_covers_slow_hello_and_stopping_is_terminal) {
}
cbm_daemon_ipc_connection_close(slow_hello);
runtime_test_fixture_finish(&fixture);
cbm_daemon_runtime_hold_peer_waits_for_testing(false);
}
ASSERT_TRUE(started);
ASSERT_TRUE(slow_slot_counted);
ASSERT_TRUE(capacity_rejected);
ASSERT_TRUE(slow_slot_released);
ASSERT_TRUE(exited);
ASSERT_TRUE(no_resurrection);
/* The daemon answers the overflow only after its HELLO is on the wire: the
* order every platform delivers deterministically. */
TEST(daemon_runtime_connection_cap_covers_slow_hello_and_stopping_is_terminal) {
runtime_cap_outcome_t outcome;
runtime_cap_scenario(RUNTIME_CAP_HELLO_BEFORE_REJECTION, &outcome);
ASSERT_TRUE(outcome.started);
ASSERT_TRUE(outcome.slow_slot_counted);
ASSERT_TRUE(outcome.order_pinned);
ASSERT_TRUE(outcome.capacity_rejected);
ASSERT_TRUE(outcome.slow_slot_released);
ASSERT_TRUE(outcome.exited);
ASSERT_TRUE(outcome.no_resurrection);
PASS();
}
#ifndef _WIN32
/* Item 12's production race, pinned: the daemon has written its capacity
* rejection AND closed before the client sends its HELLO. That send fails
* (EPIPE) while the complete rejection sits in the client's receive buffer;
* the client must still report it, not a bare transport error that bootstrap
* would classify by lock state instead of "capacity reached".
* POSIX only: a Windows server close discards unread pipe data, so there the
* rejection is delivered by the drain-before-close instead (which the other
* order covers) and this order cannot carry one by design. */
TEST(daemon_runtime_capacity_rejection_survives_daemon_close_before_hello) {
runtime_cap_outcome_t outcome;
runtime_cap_scenario(RUNTIME_CAP_REJECTION_BEFORE_HELLO, &outcome);
ASSERT_TRUE(outcome.started);
ASSERT_TRUE(outcome.slow_slot_counted);
ASSERT_TRUE(outcome.order_pinned);
ASSERT_TRUE(outcome.capacity_rejected);
ASSERT_TRUE(outcome.slow_slot_released);
ASSERT_TRUE(outcome.exited);
ASSERT_TRUE(outcome.no_resurrection);
PASS();
}
#endif
TEST(daemon_runtime_rejects_forged_identity_extension) {
cbm_daemon_build_identity_t identity =
runtime_test_identity("2.4.0", runtime_test_self_build());
@@ -5191,6 +5302,9 @@ SUITE(daemon_runtime) {
#endif
RUN_TEST(daemon_runtime_authenticated_idle_connection_outlives_lease_interval);
RUN_TEST(daemon_runtime_connection_cap_covers_slow_hello_and_stopping_is_terminal);
#ifndef _WIN32
RUN_TEST(daemon_runtime_capacity_rejection_survives_daemon_close_before_hello);
#endif
RUN_TEST(daemon_runtime_rejects_forged_identity_extension);
RUN_TEST(daemon_runtime_application_response_roundtrip_is_byte_exact);
RUN_TEST(daemon_runtime_final_disconnect_rejects_blocked_provisional_session);