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 <cursoragent@cursor.com>
This commit is contained in:
WHC
2026-10-29 13:19:03 +08:00
co-authored by Cursor
parent 2f96b050d6
commit de2eaf87f7
5 changed files with 129 additions and 45 deletions
+1
View File
@@ -24,6 +24,7 @@ typedef struct {
} cst_peer_session_t; } cst_peer_session_t;
/* Dial host:port, prove our cert, verify theirs, establish AEAD session. /* 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. */ * 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, 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], const uint8_t self_sk[64], const uint8_t realm_pk[32],
+1 -1
View File
@@ -3,7 +3,7 @@
#define CST_VERSION_MAJOR 0 #define CST_VERSION_MAJOR 0
#define CST_VERSION_MINOR 9 #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) #x
#define CST_VERSION_STR(x) CST_VERSION_STR_(x) #define CST_VERSION_STR(x) CST_VERSION_STR_(x)
+33 -16
View File
@@ -272,43 +272,59 @@ int cst_parliament_daemon_run(const char *password, const char *instance_id,
signal(SIGHUP, on_sig); signal(SIGHUP, on_sig);
#endif #endif
g_stop = 0; g_stop = 0;
int tick = 0, idle = 0; int tick = 0, idle = 0, dial_rr = 0;
while (!g_stop) { while (!g_stop) {
cst_peer_session_t sess; cst_peer_session_t sess;
char herr[128]; char herr[128];
/* Accept wait backs off when idle so a quiet seat barely wakes. */ /* Prefer accept: a blocking outbound dial used to starve inbound peers
int accept_ms = idle > 20 ? 2000 : idle > 5 ? 1000 : 400; * (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, if (cst_peer_accept((intptr_t)ls, &cert, sk, realm.pk, accept_ms, &sess,
herr, sizeof(herr)) == 0) { herr, sizeof(herr)) != 0)
break;
sync_on_session(&sess); sync_on_session(&sess);
cst_peer_close(&sess); cst_peer_close(&sess);
accepted++;
idle = 0; idle = 0;
} else { if (accepted >= 8)
idle++; break;
} }
if (accepted == 0)
idle++;
int push = cst_parliament_sync_pending(); int push = cst_parliament_sync_pending();
tick++; tick++;
/* Dial peers on push, or periodically (slower while idle). */ /* At most one outbound dial per loop turn, short wall budget, so accept
int period = idle > 20 ? 45 : idle > 5 ? 20 : 8; * 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) { if ((push || tick % period == 0) && disc) {
cst_peer_info_t peers[16]; cst_peer_info_t peers[16];
int n = cst_discover_peers(disc, peers, 16); int n = cst_discover_peers(disc, peers, 16);
if (n > 0 || push) { int tried = 0;
for (int i = 0; i < n; i++) { for (int k = 0; k < n; k++) {
int i = (dial_rr + k) % n;
if (peers[i].port <= 0) if (peers[i].port <= 0)
continue; 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, if (cst_peer_dial(peers[i].host, peers[i].port, &cert, sk, realm.pk,
3000, &sess, herr, sizeof(herr)) == 0) { 2500, &sess, herr, sizeof(herr)) == 0) {
sync_on_session(&sess); sync_on_session(&sess);
cst_peer_close(&sess); cst_peer_close(&sess);
idle = 0; idle = 0;
} }
tried = 1;
break; /* one dial then go back to accept */
} }
} if (push && (tried || n == 0))
if (push)
cst_parliament_sync_ack(); cst_parliament_sync_ack();
} }
/* No extra sleep: accept_ms already paced the loop. */
} }
if (disc) if (disc)
cst_discover_stop(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"); snprintf(err, err_sz, "service did not start");
return -1; return -1;
} }
/* Child: detach. */ /* Child: detach fully so SSH/nohup callers are not held by open stdio. */
setsid(); setsid();
int fd = open("/dev/null", O_RDWR); int fd = open("/dev/null", O_RDWR);
if (fd >= 0) { if (fd >= 0) {
dup2(fd, 0); dup2(fd, 0);
/* Keep stderr for service banner until closed by operator logs. */ dup2(fd, 1);
dup2(fd, 2);
if (fd > 2) if (fd > 2)
close(fd); close(fd);
} }
+72 -21
View File
@@ -13,6 +13,33 @@
#include <string.h> #include <string.h>
#include <time.h> #include <time.h>
#ifdef _WIN32
#include <windows.h>
#else
#include <sys/time.h>
#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 void hex_encode(const uint8_t *in, size_t n, char *out) {
static const char *H = "0123456789abcdef"; static const char *H = "0123456789abcdef";
for (size_t i = 0; i < n; i++) { 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, static int handshake_as_client(os_socket_t s, const cst_realm_cert_t *self,
const uint8_t self_sk[64], 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) { 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; return -1;
uint8_t nonce[32]; 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; 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; 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; return -1;
if (fill_random(nonce, 32) != 0) if (fill_random(nonce, 32) != 0)
return -1; 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; 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; 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 -1;
return 0; return 0;
} }
static int handshake_as_server(os_socket_t s, const cst_realm_cert_t *self, static int handshake_as_server(os_socket_t s, const cst_realm_cert_t *self,
const uint8_t self_sk[64], 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) { 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; return -1;
uint8_t nonce[32]; uint8_t nonce[32];
if (fill_random(nonce, 32) != 0) if (fill_random(nonce, 32) != 0)
return -1; 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; 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; 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; 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; 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; 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 -1;
return 0; return 0;
} }
@@ -433,15 +472,19 @@ int cst_peer_dial(const char *host, int port, const cst_realm_cert_t *self,
err[0] = 0; err[0] = 0;
if (!host || !self || !self_sk || !realm_pk || !out) if (!host || !self || !self_sk || !realm_pk || !out)
return -1; return -1;
if (timeout_ms <= 0)
timeout_ms = 3000;
int64_t deadline = peer_now_ms() + timeout_ms;
os_net_init(); 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 (s == OS_SOCKET_BAD) {
if (err && err_sz) if (err && err_sz)
snprintf(err, err_sz, "connect failed"); snprintf(err, err_sz, "connect failed");
return -1; return -1;
} }
cst_realm_cert_t remote; 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) { 0) {
if (err && err_sz) if (err && err_sz)
snprintf(err, err_sz, "peer handshake failed"); 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); snprintf(out->host, sizeof(out->host), "%s", host);
out->port = port; out->port = port;
out->sock = (intptr_t)s; 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) if (err && err_sz)
snprintf(err, err_sz, "peer key exchange failed"); snprintf(err, err_sz, "peer key exchange failed");
cst_peer_close(out); 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; err[0] = 0;
if (!self || !self_sk || !realm_pk || !out) if (!self || !self_sk || !realm_pk || !out)
return -1; return -1;
if (timeout_ms <= 0)
timeout_ms = 3000;
int timed = 0; 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); os_socket_t s = os_net_accept((os_socket_t)listen_sock, timeout_ms, &timed);
if (s == OS_SOCKET_BAD) { if (s == OS_SOCKET_BAD) {
if (err && err_sz) if (err && err_sz)
snprintf(err, err_sz, timed ? "accept timeout" : "accept failed"); snprintf(err, err_sz, timed ? "accept timeout" : "accept failed");
return -1; return -1;
} }
int64_t deadline = peer_now_ms() + (timeout_ms < 3000 ? 3000 : timeout_ms);
cst_realm_cert_t remote; 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) { 0) {
if (err && err_sz) if (err && err_sz)
snprintf(err, err_sz, "peer handshake failed"); 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)); os_net_peer(s, out->host, sizeof(out->host));
out->port = 0; out->port = 0;
out->sock = (intptr_t)s; 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) if (err && err_sz)
snprintf(err, err_sz, "peer key exchange failed"); snprintf(err, err_sz, "peer key exchange failed");
cst_peer_close(out); cst_peer_close(out);
+15
View File
@@ -9,6 +9,7 @@
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
#include <time.h>
typedef struct { typedef struct {
os_socket_t listen; os_socket_t listen;
@@ -109,8 +110,22 @@ int main(void) {
pthread_join(th, NULL); pthread_join(th, NULL);
assert(!srv2.ok); 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(ls);
os_net_close(ls2); os_net_close(ls2);
os_net_close(ls3);
os_net_cleanup(); os_net_cleanup();
puts("test_peer_auth: OK"); puts("test_peer_auth: OK");
return 0; return 0;