stratum: reap idle connections (SO_RCVTIMEO + TCP keepalive)

The recv loop previously blocked forever with no timeout, so half-open
TCPs from crashed miners and clients that authenticate but never submit
sat in ESTABLISHED state indefinitely — each holding a pthread, fd, and
per-conn state. Observed on the live pool: 22 stale sockets vs 1 active
miner.

New idle_timeout_sec config (default 600). Every accepted socket gets:

  - SO_KEEPALIVE + TCP_KEEPIDLE=120 / TCP_KEEPINTVL=30 / TCP_KEEPCNT=3
    (OS drops truly-dead TCPs after ~3.5 min of unacked probes)
  - SO_RCVTIMEO clamped to min(idle_timeout_sec, 30s) so recv() wakes
    and the loop can compare last_activity_ms against the budget

Set idle_timeout_sec < 0 to disable and get the old blocking behaviour.

Tests verify SO_RCVTIMEO is applied with the correct clamp, and that
idle_timeout_sec = 0 leaves the socket in blocking mode.

Author: rsantacroce

src/config.c

@@ -48,6 +48,7 @@ void proxy_config_defaults(proxy_config_t *cfg) {
     cfg->vardiff_min        = 1.0;
     cfg->vardiff_max        = 1e12;
     cfg->vardiff_window_sec = 30;
+    cfg->idle_timeout_sec   = 600;    /* 10 min silent recv → reap */
 
     cfg->redis_url[0] = '\0';
     cfg->redis_publish_timeout_ms   = 200;
@@ -149,6 +150,7 @@ int proxy_config_load(const char *path, proxy_config_t *cfg,
         else if (strcmp(k, "vardiff_min")               == 0) cfg->vardiff_min = atof(v);
         else if (strcmp(k, "vardiff_max")               == 0) cfg->vardiff_max = atof(v);
         else if (strcmp(k, "vardiff_window_sec")        == 0) cfg->vardiff_window_sec = atoi(v);
+        else if (strcmp(k, "idle_timeout_sec")          == 0) cfg->idle_timeout_sec = atoi(v);
         else if (strcmp(k, "db_path")                   == 0) copy_str(cfg->db_path, sizeof cfg->db_path, v);
         else if (strcmp(k, "commit_window_ms")          == 0) cfg->commit_window_ms = atoi(v);
         else if (strcmp(k, "commit_max_shares")         == 0) cfg->commit_max_shares = atoi(v);

src/config.h

@@ -19,6 +19,12 @@ typedef struct {
     double vardiff_max;           /* default 1e12; clamped by network diff */
     int    vardiff_window_sec;    /* retarget interval, default 30 */
 
+    /* Idle-connection reaper. A connection that hasn't sent any bytes in
+     * idle_timeout_sec is closed. Guards against half-open TCPs from
+     * crashed miners and clients that connect but never authenticate.
+     * Set to a negative value to disable entirely; 0 uses the default. */
+    int    idle_timeout_sec;      /* default 600 (10 min) */
+
     /* bitcoind */
     char bitcoind_url[512];
     char bitcoind_user[128];

src/main.c

@@ -533,6 +533,7 @@ int main(int argc, char **argv) {
     stcfg.vardiff_min        = cfg.vardiff_min;
     stcfg.vardiff_max        = cfg.vardiff_max;
     stcfg.vardiff_window_sec = cfg.vardiff_window_sec;
+    stcfg.idle_timeout_sec   = cfg.idle_timeout_sec;
 
     /* PPS / Thunder. Both pps and pps-classic modes require Thunder-address
      * usernames and produce PPS accruals; they differ only in the coinbase

src/stratum.c

@@ -212,6 +212,11 @@ struct stratum_conn {
     uint64_t vd_window_start_ms;
     uint32_t vd_window_shares;
 
+    /* Monotonic timestamp of the most recent recv() that got any bytes.
+     * The conn thread checks this against cfg.idle_timeout_sec after each
+     * SO_RCVTIMEO wake-up so silent connections are reaped. */
+    uint64_t last_activity_ms;
+
     pthread_mutex_t write_lock;
 
     struct stratum_conn *next;  /* server->conns_head linked list */
@@ -1160,6 +1165,53 @@ static int write_all(int fd, const char *buf, size_t len) {
     return 0;
 }
 
+/* Configure an accepted socket so idle miners get reaped instead of
+ * clogging fds/threads indefinitely. Two mechanisms, belt-and-suspenders:
+ *
+ *   1. Application-level: SO_RCVTIMEO gives recv() a bounded wake so we
+ *      can compare last_activity_ms against cfg.idle_timeout_sec. Catches
+ *      miners that hold the TCP open but never send anything (e.g. bad
+ *      username, misconfigured worker).
+ *   2. OS-level: SO_KEEPALIVE + tightened TCP_KEEPIDLE/INTVL/CNT so Linux
+ *      drops the socket after ~5 min of unacked probes. Catches half-open
+ *      TCPs where the miner box vanished from the network without FIN.
+ *
+ * idle_timeout_sec <= 0 disables the read-timeout path (legacy blocking
+ * recv). Returns 0 on success, -1 on fatal setsockopt failure. */
+static int conn_socket_setup(int fd, int idle_timeout_sec) {
+    int one = 1;
+    /* TCP_NODELAY: stratum is tiny latency-sensitive JSON, don't Nagle. */
+    (void)setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
+
+    /* Kernel keepalive. Default kernel setting is ~2h before probes even
+     * start, useless for our purposes — override with tight values. */
+    (void)setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, &one, sizeof(one));
+#ifdef TCP_KEEPIDLE
+    int idle = 120;   /* start probing after 2 min of inactivity */
+    int intvl = 30;   /* probe every 30s */
+    int cnt = 3;      /* drop after 3 unacked probes ≈ 3.5 min total */
+    (void)setsockopt(fd, IPPROTO_TCP, TCP_KEEPIDLE,  &idle,  sizeof(idle));
+    (void)setsockopt(fd, IPPROTO_TCP, TCP_KEEPINTVL, &intvl, sizeof(intvl));
+    (void)setsockopt(fd, IPPROTO_TCP, TCP_KEEPCNT,   &cnt,   sizeof(cnt));
+#endif
+
+    if (idle_timeout_sec > 0) {
+        /* Poll interval: min(idle_timeout, 30s). Longer wastes the tail
+         * of the timeout; shorter costs one recv wake per fd per interval
+         * (500 conns × wake/30s = ~16/s, negligible). */
+        int poll_s = idle_timeout_sec < 30 ? idle_timeout_sec : 30;
+        struct timeval tv = { .tv_sec = poll_s, .tv_usec = 0 };
+        if (setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)) < 0) {
+            return -1;
+        }
+    }
+    return 0;
+}
+
+int stratum_socket_setup_for_test(int fd, int idle_timeout_sec) {
+    return conn_socket_setup(fd, idle_timeout_sec);
+}
+
 static void conn_register(stratum_server_t *s, stratum_conn_t *c) {
     pthread_mutex_lock(&s->conns_lock);
     c->next = s->conns_head;
@@ -1183,9 +1235,34 @@ static void *conn_thread(void *arg) {
     char buf[MAX_LINE_BYTES + 1];
     size_t blen = 0;
 
+    /* Seed activity tracking at connect time — a client that never sends
+     * a single byte is still governed by cfg.idle_timeout_sec. */
+    c->last_activity_ms = mono_ms();
+    const uint64_t idle_timeout_ms =
+        s->cfg.idle_timeout_sec > 0
+            ? (uint64_t)s->cfg.idle_timeout_sec * 1000u
+            : 0;
+
     while (!atomic_load(&s->stop)) {
         ssize_t n = recv(c->fd, buf + blen, sizeof(buf) - 1 - blen, 0);
-        if (n <= 0) break;
+        if (n < 0) {
+            if (errno == EINTR) continue;
+            if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                /* SO_RCVTIMEO wake. Drop iff we've been silent past the
+                 * configured budget. Otherwise loop and try again. */
+                if (idle_timeout_ms > 0 &&
+                    mono_ms() - c->last_activity_ms > idle_timeout_ms) {
+                    LOG_INFO("stratum: idle timeout after %us — closing fd=%d worker='%s'",
+                             s->cfg.idle_timeout_sec, c->fd,
+                             c->worker_name[0] ? c->worker_name : "(unauthorized)");
+                    goto done;
+                }
+                continue;
+            }
+            break;  /* real socket error */
+        }
+        if (n == 0) break;  /* peer closed */
+        c->last_activity_ms = mono_ms();
         blen += (size_t)n;
         buf[blen] = '\0';
         for (;;) {
@@ -1234,8 +1311,12 @@ static void *listener_thread(void *arg) {
             close(fd);
             continue;
         }
-        int one = 1;
-        setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
+        if (conn_socket_setup(fd, s->cfg.idle_timeout_sec) < 0) {
+            LOG_WARN("stratum: socket setup failed for accepted fd: %s",
+                     strerror(errno));
+            close(fd);
+            continue;
+        }
         stratum_conn_t *c = stratum_conn_new_for_test(s);
         if (!c) { close(fd); continue; }
         c->fd = fd;
@@ -1261,6 +1342,9 @@ int stratum_server_start(const stratum_cfg_t *cfg, stratum_server_t **out) {
     s->cfg = *cfg;
     if (s->cfg.max_conns <= 0) s->cfg.max_conns = 500;
     if (s->cfg.initial_diff <= 0) s->cfg.initial_diff = 1.0;
+    /* Negative → explicit disable. 0 → apply default (10 min). Positive kept. */
+    if (s->cfg.idle_timeout_sec == 0) s->cfg.idle_timeout_sec = 600;
+    else if (s->cfg.idle_timeout_sec < 0) s->cfg.idle_timeout_sec = 0;
     pthread_rwlock_init(&s->job_lock, NULL);
     pthread_mutex_init(&s->recent_lock, NULL);
     pthread_mutex_init(&s->conns_lock, NULL);

src/stratum.h

@@ -103,6 +103,11 @@ typedef struct {
     double vardiff_max;
     int    vardiff_window_sec;
 
+    /* Drop a connection whose recv() has been silent for this long. Guards
+     * against half-open TCPs from crashed miners and misconfigured clients
+     * that connect but never authenticate. 0 disables (legacy). Default 600. */
+    int    idle_timeout_sec;
+
     void  *ctx;
     share_observer_fn  on_share;
     reject_observer_fn on_reject;
@@ -137,6 +142,12 @@ const char *stratum_conn_payout_address_for_test(const stratum_conn_t *c);
 int         stratum_conn_authorized_for_test(const stratum_conn_t *c);
 int         stratum_conn_subscribed_for_test(const stratum_conn_t *c);
 
+/* Apply the same socket options the listener applies to every accepted
+ * connection: TCP_NODELAY, SO_KEEPALIVE + TCP_KEEP{IDLE,INTVL,CNT}, and
+ * SO_RCVTIMEO (poll interval derived from idle_timeout_sec). Exposed for
+ * tests. */
+int stratum_socket_setup_for_test(int fd, int idle_timeout_sec);
+
 /* Process one JSON-RPC line. Appends one or more newline-delimited JSON
  * messages to *out_buf (caller-owned, will be realloc'd). Returns 0 on
  * success, negative on protocol error (caller should disconnect). */

tests/test_stratum.c

@@ -6,6 +6,9 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <sys/socket.h>
+#include <sys/time.h>
+#include <unistd.h>
 
 static int g_pass = 0;
 static int g_fail = 0;
@@ -280,13 +283,52 @@ static void test_authorize_address_with_label(void) {
     stratum_server_free(s);
 }
 
+/* Idle-socket reaper: verify the accepted-socket setup path applies
+ * SO_RCVTIMEO derived from idle_timeout_sec. We can't cheaply test the
+ * "silent client gets dropped" path in a unit test — that would need a
+ * real 3s+ sleep — but confirming SO_RCVTIMEO is present is enough to
+ * guarantee the recv loop wakes and gets to check last_activity_ms. */
+static void test_socket_setup_applies_rcvtimeo(void) {
+    int sp[2];
+    CHECK(socketpair(AF_UNIX, SOCK_STREAM, 0, sp) == 0);
+    /* idle_timeout=45 → poll interval clamped to 30s (SO_RCVTIMEO ceiling). */
+    CHECK(stratum_socket_setup_for_test(sp[0], 45) == 0);
+    struct timeval tv = {0};
+    socklen_t len = sizeof tv;
+    CHECK(getsockopt(sp[0], SOL_SOCKET, SO_RCVTIMEO, &tv, &len) == 0);
+    CHECK(tv.tv_sec == 30);
+    /* idle_timeout=5 → poll interval matches (below the 30s clamp). */
+    CHECK(stratum_socket_setup_for_test(sp[1], 5) == 0);
+    struct timeval tv2 = {0};
+    socklen_t len2 = sizeof tv2;
+    CHECK(getsockopt(sp[1], SOL_SOCKET, SO_RCVTIMEO, &tv2, &len2) == 0);
+    CHECK(tv2.tv_sec == 5);
+    close(sp[0]);
+    close(sp[1]);
+}
+
+/* idle_timeout_sec=0 leaves SO_RCVTIMEO unset — legacy blocking recv. */
+static void test_socket_setup_disabled(void) {
+    int sp[2];
+    CHECK(socketpair(AF_UNIX, SOCK_STREAM, 0, sp) == 0);
+    CHECK(stratum_socket_setup_for_test(sp[0], 0) == 0);
+    struct timeval tv = {0};
+    socklen_t len = sizeof tv;
+    CHECK(getsockopt(sp[0], SOL_SOCKET, SO_RCVTIMEO, &tv, &len) == 0);
+    CHECK(tv.tv_sec == 0 && tv.tv_usec == 0);
+    close(sp[0]);
+    close(sp[1]);
+}
+
 int main(void) {
     test_subscribe();
     test_authorize_triggers_setdiff_notify();
     test_submit_unknown_job();
     test_submit_share_and_dedupe();
     test_authorize_rejects_non_address();
     test_authorize_address_with_label();
+    test_socket_setup_applies_rcvtimeo();
+    test_socket_setup_disabled();
     printf("test_stratum: %d passed, %d failed\n", g_pass, g_fail);
     return g_fail == 0 ? 0 : 1;
 }