[Bug]: Linux intermittent SIGPIPE crashes while proxying traffic #268

diff --git a/Linux/src/pb_relay.c b/Linux/src/pb_relay.c
index cd51797..9c48a99 100644
--- a/Linux/src/pb_relay.c
+++ b/Linux/src/pb_relay.c
@@ -11,6 +11,7 @@
 #include <netinet/tcp.h>
 #include <poll.h>
 #include <pthread.h>
+#include <signal.h>
 #include <stdarg.h>
 #include <stdatomic.h>
 #include <stdio.h>
@@ -1166,9 +1167,18 @@ static void free_grave(struct dthread *t)
     while (t->grave) { struct conn *n = t->grave->next; free(t->grave); t->grave = n; }
 }
 
+// splice() has no MSG_NOSIGNAL equivalent. Block SIGPIPE only on the private
+// data threads, for their entire lifetime (pending signals die with the thread).
+// Do not change the embedding application's process-wide signal disposition.
 static void *data_main(void *arg)
 {
     struct dthread *t = arg;
+    sigset_t set;
+    sigemptyset(&set);
+    sigaddset(&set, SIGPIPE);
+    if (pthread_sigmask(SIG_BLOCK, &set, NULL) != 0)
+        t->nosplice = true; // safe copy path already uses MSG_NOSIGNAL
+
     struct epoll_event evs[64];
     int64_t last_sweep = now_ms();
     int64_t sweep_every = (int64_t)R.cfg.tcp_idle_timeout_s * 250;      // idle/4, clamped to 1s..5s
diff --git a/Linux/src/tests/run_relay_test.sh b/Linux/src/tests/run_relay_test.sh
index c9cf79d..ed61e89 100755
--- a/Linux/src/tests/run_relay_test.sh
+++ b/Linux/src/tests/run_relay_test.sh
@@ -3,6 +3,7 @@
 set -e
 cd "$(dirname "$0")"
 mode="${1:-both}"
+sh ./run_sigpipe_test.sh
 out="${TMPDIR:-/tmp}"
 CF="-std=gnu11 -Wall -Wextra -O1 -g -D_GNU_SOURCE -fstack-protector-strong -I.. "
 if [ "$mode" = plain ] || [ "$mode" = both ]; then
diff --git a/Linux/src/tests/sigpipe_test.c b/Linux/src/tests/sigpipe_test.c
new file mode 100644
index 0000000..f2abecc
--- /dev/null
+++ b/Linux/src/tests/sigpipe_test.c
@@ -0,0 +1,130 @@
+// Isolated white-box regression: exercise the actual static relay functions.
+#include "../pb_relay.c"
+#include <assert.h>
+#include <sys/wait.h>
+
+// The stopped data_main path does not observe DNS. Fail if it ever does.
+void pb_core_dns_observe(const uint8_t *msg, size_t len)
+{
+    (void)msg; (void)len;
+    assert(0);
+}
+
+static int tcp_mode;
+
+static void pair(int fd[2])
+{
+    if (!tcp_mode) {
+        assert(socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK, 0, fd) == 0);
+        return;
+    }
+    int listener = socket(AF_INET, SOCK_STREAM, 0);
+    assert(listener >= 0);
+    struct sockaddr_in addr = { .sin_family = AF_INET, .sin_addr.s_addr = htonl(INADDR_LOOPBACK) };
+    assert(bind(listener, (struct sockaddr *)&addr, sizeof addr) == 0);
+    socklen_t len = sizeof addr;
+    assert(getsockname(listener, (struct sockaddr *)&addr, &len) == 0);
+    assert(listen(listener, 1) == 0);
+    fd[1] = socket(AF_INET, SOCK_STREAM, 0); assert(fd[1] >= 0);
+    assert(connect(fd[1], (struct sockaddr *)&addr, len) == 0);
+    fd[0] = accept4(listener, NULL, NULL, SOCK_NONBLOCK); assert(fd[0] >= 0);
+    close(listener);
+}
+
+static void exercise(int direction)
+{
+    struct dthread t = {0};
+    struct conn c = {0};
+    int src[2], dst[2];
+    pair(src); pair(dst);
+    assert(pipe2(t.pfd, O_NONBLOCK) == 0);
+    c.fd[direction] = src[0]; c.fd[1 - direction] = dst[0];
+    if (tcp_mode) {
+        struct linger rst = { .l_onoff = 1, .l_linger = 0 };
+        assert(setsockopt(dst[1], SOL_SOCKET, SO_LINGER, &rst, sizeof rst) == 0);
+    }
+    close(dst[1]);
+    if (tcp_mode) {
+        // Synchronize the reset and consume ECONNRESET: the next write is EPIPE.
+        struct pollfd p = { .fd = dst[0], .events = POLLIN };
+        assert(poll(&p, 1, 1000) == 1);
+        char byte;
+        assert(recv(dst[0], &byte, 1, 0) == -1 && errno == ECONNRESET);
+    }
+    assert(send(src[1], "broken", 6, MSG_NOSIGNAL) == 6);
+    assert(!t.nosplice);
+    assert(pump_splice(&t, &c, direction) == -1);
+    sigset_t pending;
+    assert(sigpending(&pending) == 0 && sigismember(&pending, SIGPIPE) == 1);
+    char b[8];
+    assert(read(t.pfd[0], b, sizeof b) == -1 && errno == EAGAIN);
+    close(src[0]); close(src[1]); close(dst[0]);
+
+    // Another connection on the same worker and reusable pipe still works.
+    pair(src); pair(dst);
+    c.fd[direction] = src[0]; c.fd[1 - direction] = dst[0];
+    assert(send(src[1], "alive", 5, MSG_NOSIGNAL) == 5);
+    assert(pump_splice(&t, &c, direction) == 1);
+    assert(recv(dst[1], b, sizeof b, 0) == 5 && memcmp(b, "alive", 5) == 0);
+    close(src[0]); close(src[1]); close(dst[0]); close(dst[1]);
+    close(t.pfd[0]); close(t.pfd[1]);
+}
+
+static void *protected_worker(void *arg)
+{
+    // Run the production thread entry through its shutdown path, then exercise
+    // writes before this private thread exits, with its resulting signal mask.
+    struct dthread t = {0};
+    atomic_store(&R.stopping, true);
+    data_main(&t);
+    exercise(*(int *)arg);
+    return NULL;
+}
+
+static void negotiations(void)
+{
+    int fd[2]; pair(fd); close(fd[1]);
+    struct io x = { .fd = fd[0], .stopfd = -1, .dl = now_ms() + 1000 };
+    struct pb_proxy p = {0};
+    struct s5dst d = { .atyp = 1, .port = 80 };
+    uint8_t left[HDR_MAX]; size_t n = 0;
+    assert(socks5_request(&x, &p, 1, &d, NULL) == -1 && errno == EPIPE);
+    assert(socks5_request(&x, &p, 3, &d, NULL) == -1 && errno == EPIPE);
+    assert(http_connect(&x, &p, "localhost", 80, left, &n) == -1 && errno == EPIPE);
+    close(fd[0]);
+}
+
+int main(void)
+{
+    // Never depend on the shell/Python parent's inherited SIGPIPE policy.
+    assert(signal(SIGPIPE, SIG_DFL) != SIG_ERR);
+    sigset_t set; sigemptyset(&set); sigaddset(&set, SIGPIPE);
+    assert(pthread_sigmask(SIG_UNBLOCK, &set, NULL) == 0);
+    for (tcp_mode = 0; tcp_mode < 2; tcp_mode++) {
+        for (int d = 0; d < 2; d++) {
+            for (int protect = 0; protect < 2; protect++) {
+                pid_t pid = fork(); assert(pid >= 0);
+                if (!pid) {
+                    alarm(5);
+                    if (protect) {
+                        pthread_t th;
+                        assert(pthread_create(&th, NULL, protected_worker, &d) == 0);
+                        assert(pthread_join(th, NULL) == 0);
+                        sigset_t mask;
+                        assert(pthread_sigmask(SIG_BLOCK, NULL, &mask) == 0);
+                        assert(sigismember(&mask, SIGPIPE) == 0);
+                    } else exercise(d);
+                    _exit(0);
+                }
+                int st; assert(waitpid(pid, &st, 0) == pid);
+                if (protect) assert(WIFEXITED(st) && WEXITSTATUS(st) == 0);
+                else assert(WIFSIGNALED(st) && WTERMSIG(st) == SIGPIPE);
+            }
+        }
+    }
+    tcp_mode = 0;
+    negotiations(); // unblocked SIGPIPE: MSG_NOSIGNAL protects existing paths
+    struct sigaction sa; assert(sigaction(SIGPIPE, NULL, &sa) == 0);
+    assert(sa.sa_handler == SIG_DFL);
+    puts("PASS: UNIX/TCP raw splice SIGPIPE reproduced; protected forwarding survives in both directions; SOCKS5 TCP/UDP and HTTP negotiation return EPIPE");
+}
diff --git a/Linux/src/tests/run_sigpipe_test.sh b/Linux/src/tests/run_sigpipe_test.sh
new file mode 100755
index 0000000..97fb6f4
--- /dev/null
+++ b/Linux/src/tests/run_sigpipe_test.sh
@@ -0,0 +1,9 @@
+#!/bin/sh
+set -eu
+cd "$(dirname "$0")"
+work=$(mktemp -d)
+trap 'rm -rf "$work"' EXIT
+${CC:-cc} -std=gnu11 -D_GNU_SOURCE -O2 -g -Wall -Wextra -Werror \
+    -ffunction-sections -fdata-sections sigpipe_test.c -Wl,--gc-sections \
+    -pthread -o "$work/sigpipe_test"
+"$work/sigpipe_test"
diff --git a/Linux/src/tests/SIGPIPE.md b/Linux/src/tests/SIGPIPE.md
new file mode 100644
index 0000000..f445503
--- /dev/null
+++ b/Linux/src/tests/SIGPIPE.md
@@ -0,0 +1,81 @@
+# SIGPIPE regression
+
+## Confirmed mechanism
+
+`pb_relay.c:pump_splice()` forwards TCP bytes through a reusable pipe. Its
+pipe-to-socket `splice()` can raise SIGPIPE when the destination connection has
+broken. There is no `MSG_NOSIGNAL` flag for `splice()`. With the default signal
+disposition this terminates the entire process, before the existing fatal-error
+path can discard the pipe and close the affected connection.
+
+The path is `data_main -> conn_service -> conn_pump -> pump_splice`, in either
+direction, after SOCKS5/HTTP negotiation or for DIRECT TCP. A TCP reset may first
+report ECONNRESET; subsequent output can report EPIPE and raise SIGPIPE. Readiness
+notifications do not guarantee that the peer remains connected until the write.
+
+Ordinary socket sends already use MSG_NOSIGNAL: `io_send` (SOCKS5 CONNECT,
+authentication, UDP ASSOCIATE, HTTP CONNECT, proxy checks), the TCP copy/pending
+and DNS paths, and UDP send/sendto/sendmsg. Negotiation handles partial writes
+and EINTR/EAGAIN; fatal send/receive errors fail that handshake. The TCP pump
+buffers partial writes and closes only that connection on fatal errors.
+Connection descriptors are owned by the data thread after worker handoff; the
+SIGPIPE failure does not require a concurrent close or descriptor reuse race.
+`eventfd_write` targets eventfds, and profile writes target temporary files;
+these are not broken stream/pipe writes.
+
+The fix blocks SIGPIPE for the entire lifetime of each private TCP data thread.
+Synchronous SIGPIPE from splice stays pending on that thread and disappears at
+thread exit. EPIPE/ECONNRESET still follow the existing connection error path.
+No process-wide disposition is changed, no per-chunk syscalls are added, and
+zero-copy remains enabled. If signal masking fails, the thread uses the existing
+MSG_NOSIGNAL copy path. That exceptional fallback is not fault-injection tested.
+
+## Tests
+
+Run `sh Linux/src/tests/run_sigpipe_test.sh`, or the existing
+`sh Linux/src/tests/run_relay_test.sh both`, which includes it.
+
+The isolated test calls the actual static relay functions, with unrelated
+sections discarded at link time. It explicitly restores SIG_DFL and unblocks
+SIGPIPE, so an inherited Python/shell workaround cannot hide a regression.
+It uses UNIX socketpairs and real loopback TCP connections. TCP peers close
+with SO_LINGER reset; the test consumes ECONNRESET before forwarding to make
+EPIPE deterministic. For both directions it verifies:
+
+- unprotected pump_splice terminates a child with signal 13;
+- the production data_main signal setup prevents that termination;
+- SIGPIPE is pending, the fatal splice returns, and the pipe is empty;
+- another connection transfers intact bytes using the same thread and pipe;
+- the initiating thread's signal mask and process disposition stay unchanged.
+
+With SIGPIPE unblocked, SOCKS5 CONNECT, UDP ASSOCIATE and HTTP CONNECT also
+return EPIPE on a broken socket instead of terminating the process.
+
+## Attribution and limits
+
+This reproduces a real vulnerability in the relay, but does not identify the
+syscall responsible for a previously observed exit 141 without its stack or
+trace. CLI log/connection callbacks in `Linux/cli/main.c` also use printf to
+stdout: a closed output pipe can independently cause SIGPIPE. This patch does
+not change CLI broken-output semantics or signal policy in an embedding app.
+
+The test does not load eBPF or change routes. Live eBPF integration and the
+original multi-application workload require separate runtime acceptance.
+
+## Validation in this checkout
+
+- Dedicated regression: passes with the patch; compiling the same test against
+  the original HEAD source fails when the nominally protected worker gets SIGPIPE.
+- `PB_CHURN_TCP=10000 PB_CHURN_UDP=3000 sh Linux/src/tests/run_relay_test.sh both`:
+  plain 133 passed / 0 failed; ASan+UBSan 133 passed / 0 failed, exit status 0.
+  Each run completed 8 churn chunks (8000 TCP + 2400 UDP sessions) within the
+  harness's chunk/budget policy. The suite also exercised 10000 idle TCP connections.
+- `make -C Linux/src BUILD_DIR=/tmp/proxybridge-sigpipe-build test-rules`:
+  rules/DNS passed, proc 18 checks / 0 failed, with ASan+UBSan.
+- `pb_relay.o` builds using the production hardening flags and WERROR=1.
+  Full library/eBPF build cannot finish: system `bpf/bpf_helpers.h` is missing.
+- Shell syntax checks and `git diff --check` pass.
+
+Socket tests and LeakSanitizer required execution outside the restricted sandbox
+(send returned EPERM and LeakSanitizer rejected ptrace there). No live engine,
+system routing changes, installation, commit or push was performed.