From de2eaf87f7387907997330905f8dd5b7703cc91c Mon Sep 17 00:00:00 2001 From: WHC Date: Thu, 29 Oct 2026 13:19:03 +0800 Subject: [PATCH] fix: stop parliament accept starvation on outbound dial (0.9.22) Peer dial used per-step read timeouts so a stuck handshake could block the single-threaded daemon for tens of seconds; inbound hellos piled up unread (CLOSE-WAIT). Use a wall-clock dial budget, drain accepts before each dial, skip self, and only one side initiates idle polls. Co-authored-by: Cursor --- include/cst_peer.h | 1 + include/cst_version.h | 2 +- src/parliament_daemon.c | 63 ++++++++++++++++++---------- src/peer_auth.c | 93 +++++++++++++++++++++++++++++++---------- tests/test_peer_auth.c | 15 +++++++ 5 files changed, 129 insertions(+), 45 deletions(-) diff --git a/include/cst_peer.h b/include/cst_peer.h index 812bbb5..618a9e9 100644 --- a/include/cst_peer.h +++ b/include/cst_peer.h @@ -24,6 +24,7 @@ typedef struct { } cst_peer_session_t; /* Dial host:port, prove our cert, verify theirs, establish AEAD session. + * timeout_ms is a wall-clock budget for connect+handshake+kx (not per-read). * Leaves sock open for post-handshake sync; call cst_peer_close. */ int cst_peer_dial(const char *host, int port, const cst_realm_cert_t *self, const uint8_t self_sk[64], const uint8_t realm_pk[32], diff --git a/include/cst_version.h b/include/cst_version.h index c389f18..68ae0e5 100644 --- a/include/cst_version.h +++ b/include/cst_version.h @@ -3,7 +3,7 @@ #define CST_VERSION_MAJOR 0 #define CST_VERSION_MINOR 9 -#define CST_VERSION_PATCH 20 +#define CST_VERSION_PATCH 22 #define CST_VERSION_STR_(x) #x #define CST_VERSION_STR(x) CST_VERSION_STR_(x) diff --git a/src/parliament_daemon.c b/src/parliament_daemon.c index 8350a47..77bb237 100644 --- a/src/parliament_daemon.c +++ b/src/parliament_daemon.c @@ -272,43 +272,59 @@ int cst_parliament_daemon_run(const char *password, const char *instance_id, signal(SIGHUP, on_sig); #endif g_stop = 0; - int tick = 0, idle = 0; + int tick = 0, idle = 0, dial_rr = 0; while (!g_stop) { cst_peer_session_t sess; char herr[128]; - /* Accept wait backs off when idle so a quiet seat barely wakes. */ - int accept_ms = idle > 20 ? 2000 : idle > 5 ? 1000 : 400; - if (cst_peer_accept((intptr_t)ls, &cert, sk, realm.pk, accept_ms, &sess, - herr, sizeof(herr)) == 0) { + /* Prefer accept: a blocking outbound dial used to starve inbound peers + * (CLOSE-WAIT + unread hello ~385B on :9100). Drain the backlog first. */ + int accepted = 0; + for (;;) { + int accept_ms = accepted > 0 ? 50 : (idle > 20 ? 800 : idle > 5 ? 400 : 200); + if (cst_peer_accept((intptr_t)ls, &cert, sk, realm.pk, accept_ms, &sess, + herr, sizeof(herr)) != 0) + break; sync_on_session(&sess); cst_peer_close(&sess); + accepted++; idle = 0; - } else { - idle++; + if (accepted >= 8) + break; } + if (accepted == 0) + idle++; + int push = cst_parliament_sync_pending(); tick++; - /* Dial peers on push, or periodically (slower while idle). */ - int period = idle > 20 ? 45 : idle > 5 ? 20 : 8; + /* At most one outbound dial per loop turn, short wall budget, so accept + * keeps pace. Lexicographic rule avoids A↔B mutual-dial deadlock unless + * we have a local push to deliver. */ + int period = idle > 20 ? 30 : idle > 5 ? 12 : 6; if ((push || tick % period == 0) && disc) { cst_peer_info_t peers[16]; int n = cst_discover_peers(disc, peers, 16); - if (n > 0 || push) { - for (int i = 0; i < n; i++) { - if (peers[i].port <= 0) - continue; - if (cst_peer_dial(peers[i].host, peers[i].port, &cert, sk, realm.pk, - 3000, &sess, herr, sizeof(herr)) == 0) { - sync_on_session(&sess); - cst_peer_close(&sess); - idle = 0; - } + int tried = 0; + for (int k = 0; k < n; k++) { + int i = (dial_rr + k) % n; + if (peers[i].port <= 0) + continue; + if (!strcmp(peers[i].instance_id, instance_id)) + continue; /* never dial ourselves */ + if (!push && strcmp(instance_id, peers[i].instance_id) > 0) + continue; /* lower id initiates idle polls */ + dial_rr = (i + 1) % n; + if (cst_peer_dial(peers[i].host, peers[i].port, &cert, sk, realm.pk, + 2500, &sess, herr, sizeof(herr)) == 0) { + sync_on_session(&sess); + cst_peer_close(&sess); + idle = 0; } + tried = 1; + break; /* one dial then go back to accept */ } - if (push) + if (push && (tried || n == 0)) cst_parliament_sync_ack(); } - /* No extra sleep: accept_ms already paced the loop. */ } if (disc) cst_discover_stop(disc); @@ -401,12 +417,13 @@ int cst_parliament_daemon_start(const char *password, const char *instance_id, snprintf(err, err_sz, "service did not start"); return -1; } - /* Child: detach. */ + /* Child: detach fully so SSH/nohup callers are not held by open stdio. */ setsid(); int fd = open("/dev/null", O_RDWR); if (fd >= 0) { dup2(fd, 0); - /* Keep stderr for service banner until closed by operator logs. */ + dup2(fd, 1); + dup2(fd, 2); if (fd > 2) close(fd); } diff --git a/src/peer_auth.c b/src/peer_auth.c index d6bab2b..cc83233 100644 --- a/src/peer_auth.c +++ b/src/peer_auth.c @@ -13,6 +13,33 @@ #include #include +#ifdef _WIN32 +#include +#else +#include +#endif + +/* Wall-clock deadline so a multi-step handshake cannot stretch to + * N_steps * timeout_ms (that starved the daemon accept loop). */ +static int64_t peer_now_ms(void) { +#ifdef _WIN32 + return (int64_t)GetTickCount64(); +#else + struct timeval tv; + gettimeofday(&tv, NULL); + return (int64_t)tv.tv_sec * 1000 + (int64_t)tv.tv_usec / 1000; +#endif +} + +static int peer_remain_ms(int64_t deadline_ms) { + int64_t left = deadline_ms - peer_now_ms(); + if (left <= 0) + return 0; + if (left > 600000) + return 600000; + return (int)left; +} + static void hex_encode(const uint8_t *in, size_t n, char *out) { static const char *H = "0123456789abcdef"; for (size_t i = 0; i < n; i++) { @@ -379,48 +406,60 @@ static int recv_sealed(cst_peer_session_t *sess, uint8_t **out, size_t *out_n, static int handshake_as_client(os_socket_t s, const cst_realm_cert_t *self, const uint8_t self_sk[64], - const uint8_t realm_pk[32], int timeout_ms, + const uint8_t realm_pk[32], int64_t deadline_ms, cst_realm_cert_t *remote) { - if (send_hello(s, self, timeout_ms) != 0) + int t; + if ((t = peer_remain_ms(deadline_ms)) <= 0 || send_hello(s, self, t) != 0) return -1; uint8_t nonce[32]; - if (read_chal(s, timeout_ms, nonce) != 0) + if ((t = peer_remain_ms(deadline_ms)) <= 0 || read_chal(s, t, nonce) != 0) return -1; - if (send_auth(s, self_sk, nonce, timeout_ms) != 0) + if ((t = peer_remain_ms(deadline_ms)) <= 0 || + send_auth(s, self_sk, nonce, t) != 0) return -1; - if (read_hello(s, realm_pk, timeout_ms, remote) != 0) + if ((t = peer_remain_ms(deadline_ms)) <= 0 || + read_hello(s, realm_pk, t, remote) != 0) return -1; if (fill_random(nonce, 32) != 0) return -1; - if (send_chal(s, nonce, timeout_ms) != 0) + if ((t = peer_remain_ms(deadline_ms)) <= 0 || send_chal(s, nonce, t) != 0) return -1; - if (read_auth(s, remote, nonce, timeout_ms) != 0) + if ((t = peer_remain_ms(deadline_ms)) <= 0 || + read_auth(s, remote, nonce, t) != 0) return -1; - if (send_ok(s, timeout_ms) != 0 || read_ok(s, timeout_ms) != 0) + if ((t = peer_remain_ms(deadline_ms)) <= 0 || send_ok(s, t) != 0) + return -1; + if ((t = peer_remain_ms(deadline_ms)) <= 0 || read_ok(s, t) != 0) return -1; return 0; } static int handshake_as_server(os_socket_t s, const cst_realm_cert_t *self, const uint8_t self_sk[64], - const uint8_t realm_pk[32], int timeout_ms, + const uint8_t realm_pk[32], int64_t deadline_ms, cst_realm_cert_t *remote) { - if (read_hello(s, realm_pk, timeout_ms, remote) != 0) + int t; + if ((t = peer_remain_ms(deadline_ms)) <= 0 || + read_hello(s, realm_pk, t, remote) != 0) return -1; uint8_t nonce[32]; if (fill_random(nonce, 32) != 0) return -1; - if (send_chal(s, nonce, timeout_ms) != 0) + if ((t = peer_remain_ms(deadline_ms)) <= 0 || send_chal(s, nonce, t) != 0) return -1; - if (read_auth(s, remote, nonce, timeout_ms) != 0) + if ((t = peer_remain_ms(deadline_ms)) <= 0 || + read_auth(s, remote, nonce, t) != 0) return -1; - if (send_hello(s, self, timeout_ms) != 0) + if ((t = peer_remain_ms(deadline_ms)) <= 0 || send_hello(s, self, t) != 0) return -1; - if (read_chal(s, timeout_ms, nonce) != 0) + if ((t = peer_remain_ms(deadline_ms)) <= 0 || read_chal(s, t, nonce) != 0) return -1; - if (send_auth(s, self_sk, nonce, timeout_ms) != 0) + if ((t = peer_remain_ms(deadline_ms)) <= 0 || + send_auth(s, self_sk, nonce, t) != 0) return -1; - if (read_ok(s, timeout_ms) != 0 || send_ok(s, timeout_ms) != 0) + if ((t = peer_remain_ms(deadline_ms)) <= 0 || read_ok(s, t) != 0) + return -1; + if ((t = peer_remain_ms(deadline_ms)) <= 0 || send_ok(s, t) != 0) return -1; return 0; } @@ -433,15 +472,19 @@ int cst_peer_dial(const char *host, int port, const cst_realm_cert_t *self, err[0] = 0; if (!host || !self || !self_sk || !realm_pk || !out) return -1; + if (timeout_ms <= 0) + timeout_ms = 3000; + int64_t deadline = peer_now_ms() + timeout_ms; os_net_init(); - os_socket_t s = os_net_connect(host, port, timeout_ms); + int t = peer_remain_ms(deadline); + os_socket_t s = os_net_connect(host, port, t); if (s == OS_SOCKET_BAD) { if (err && err_sz) snprintf(err, err_sz, "connect failed"); return -1; } cst_realm_cert_t remote; - if (handshake_as_client(s, self, self_sk, realm_pk, timeout_ms, &remote) != + if (handshake_as_client(s, self, self_sk, realm_pk, deadline, &remote) != 0) { if (err && err_sz) snprintf(err, err_sz, "peer handshake failed"); @@ -453,7 +496,8 @@ int cst_peer_dial(const char *host, int port, const cst_realm_cert_t *self, snprintf(out->host, sizeof(out->host), "%s", host); out->port = port; out->sock = (intptr_t)s; - if (peer_kx(s, 1, out->aead_key, timeout_ms) != 0) { + t = peer_remain_ms(deadline); + if (t <= 0 || peer_kx(s, 1, out->aead_key, t) != 0) { if (err && err_sz) snprintf(err, err_sz, "peer key exchange failed"); cst_peer_close(out); @@ -471,15 +515,21 @@ int cst_peer_accept(intptr_t listen_sock, const cst_realm_cert_t *self, err[0] = 0; if (!self || !self_sk || !realm_pk || !out) return -1; + if (timeout_ms <= 0) + timeout_ms = 3000; int timed = 0; + /* Accept wait uses the caller budget; handshake+kx share a fresh deadline + * after the connection is taken, so a long accept wait does not shrink the + * crypto exchange to zero. */ os_socket_t s = os_net_accept((os_socket_t)listen_sock, timeout_ms, &timed); if (s == OS_SOCKET_BAD) { if (err && err_sz) snprintf(err, err_sz, timed ? "accept timeout" : "accept failed"); return -1; } + int64_t deadline = peer_now_ms() + (timeout_ms < 3000 ? 3000 : timeout_ms); cst_realm_cert_t remote; - if (handshake_as_server(s, self, self_sk, realm_pk, timeout_ms, &remote) != + if (handshake_as_server(s, self, self_sk, realm_pk, deadline, &remote) != 0) { if (err && err_sz) snprintf(err, err_sz, "peer handshake failed"); @@ -491,7 +541,8 @@ int cst_peer_accept(intptr_t listen_sock, const cst_realm_cert_t *self, os_net_peer(s, out->host, sizeof(out->host)); out->port = 0; out->sock = (intptr_t)s; - if (peer_kx(s, 0, out->aead_key, timeout_ms) != 0) { + int t = peer_remain_ms(deadline); + if (t <= 0 || peer_kx(s, 0, out->aead_key, t) != 0) { if (err && err_sz) snprintf(err, err_sz, "peer key exchange failed"); cst_peer_close(out); diff --git a/tests/test_peer_auth.c b/tests/test_peer_auth.c index dfcae6a..9a6da9c 100644 --- a/tests/test_peer_auth.c +++ b/tests/test_peer_auth.c @@ -9,6 +9,7 @@ #include #include #include +#include typedef struct { os_socket_t listen; @@ -109,8 +110,22 @@ int main(void) { pthread_join(th, NULL); assert(!srv2.ok); + /* Wall-clock dial budget: listener that never accepts must fail fast. */ + os_socket_t ls3 = os_net_listen("127.0.0.1", 0, err, sizeof(err)); + assert(ls3 != OS_SOCKET_BAD); + int port3 = os_net_port(ls3); + struct timespec t0, t1; + clock_gettime(CLOCK_MONOTONIC, &t0); + assert(cst_peer_dial("127.0.0.1", port3, &ca, ska, realm.pk, 800, &sess, err, + sizeof(err)) != 0); + clock_gettime(CLOCK_MONOTONIC, &t1); + long ms = (long)((t1.tv_sec - t0.tv_sec) * 1000 + + (t1.tv_nsec - t0.tv_nsec) / 1000000); + assert(ms >= 400 && ms < 2500); + os_net_close(ls); os_net_close(ls2); + os_net_close(ls3); os_net_cleanup(); puts("test_peer_auth: OK"); return 0;