commit d31c0a157206dfad225ac7ec8b401547b6f70d79
| author | 斟酌 鵬兄 <tgckpg@gmail.com> |
|---|
| date | 2026-05-29T06:19:01Z |
|---|
| subject | stream: drain client output before closing on upstream EOF/error |
|---|
commit d31c0a157206dfad225ac7ec8b401547b6f70d79
Author: 斟酌 鵬兄 <tgckpg@gmail.com>
Date: 2026-05-29T06:19:01Z
stream: drain client output before closing on upstream EOF/error
Rule chaining exposed a stream lifetime bug:
tcp -> unix -> file
The downstream route could receive and queue the file response, then see EOF
or EPIPE from the upstream side while request bytes were still pending toward
that upstream. The old event path treated upstream EOF/error as fatal for the
whole connection and freed both bufferevents immediately. That could discard
already-queued client output before it was flushed.
Handle this as a half-close instead:
- if upstream EOF/error arrives while client output is pending, stop the
upstream side and let the client output drain
- after the client output buffer drains, shut down the client write side
- treat the expected client EOF/reset after that as normal cleanup
This fixes chained stream routes where a server-speaks-first backend, such as
file://, responds before consuming the full client request.
---
src/stream_conn.c | 183 ++++++++++++++++++++++++++++--
src/stream_route.h | 3 +
tests/test_unix_listeners.py | 257 ++++++++++++++++++++-----------------------
3 files changed, 294 insertions(+), 149 deletions(-)
diff --git a/src/stream_conn.c b/src/stream_conn.c
index c78c03a..5c3701a 100644
--- a/src/stream_conn.c
+++ b/src/stream_conn.c
@@ -70,7 +70,48 @@ static int set_socket_keepalive(evutil_socket_t fd, const struct route *r)
return 0;
}
-void event_cb(struct bufferevent *bev, short events, void *arg) {
+#ifdef TINYPROXY_DEBUG
+static const char *bev_side(conn_t *conn, struct bufferevent *bev)
+{
+ if (bev == conn->client) {
+ return "client";
+ }
+ if (bev == conn->upstream) {
+ return "upstream";
+ }
+ return "unknown";
+}
+#endif
+
+static size_t bev_output_len(struct bufferevent *bev)
+{
+ if (bev == NULL) {
+ return 0;
+ }
+
+ return evbuffer_get_length(bufferevent_get_output(bev));
+}
+
+static bool client_has_pending_output(conn_t *conn)
+{
+ return conn->client != NULL && bev_output_len(conn->client) > 0;
+}
+
+static void drain_client_then_close(conn_t *conn)
+{
+ conn->close_client_after_drain = true;
+
+ if (conn->upstream != NULL) {
+ bufferevent_disable(conn->upstream, EV_READ | EV_WRITE);
+ }
+
+ if (!client_has_pending_output(conn)) {
+ free_conn(conn);
+ }
+}
+
+void event_cb(struct bufferevent *bev, short events, void *arg)
+{
conn_t *conn = arg;
if (events & BEV_EVENT_CONNECTED) {
@@ -87,18 +128,54 @@ void event_cb(struct bufferevent *bev, short events, void *arg) {
return;
}
- if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR)) {
- if (events & BEV_EVENT_ERROR) {
- int err = EVUTIL_SOCKET_ERROR();
- LOG_ERROR("connection error",
- "err", _LOGV(evutil_socket_error_to_string(err))
+ if (events & BEV_EVENT_ERROR) {
+ int err = EVUTIL_SOCKET_ERROR();
+
+ if (bev == conn->client &&
+ conn->close_after_client_eof &&
+ err == ECONNRESET) {
+ LOG_INFO("client reset after response drain");
+ free_conn(conn);
+ return;
+ }
+
+ LOG_ERROR("connection error",
+ "err", _LOGV(evutil_socket_error_to_string(err))
+ );
+
+ if (bev == conn->upstream && client_has_pending_output(conn)) {
+ const struct route *r = conn->route;
+
+ LOG_WARN("upstream error after response queued; draining client",
+ "err", _LOGV(evutil_socket_error_to_string(err)),
+ "client_output", _LOGV(bev_output_len(conn->client)),
+ "upstream_output", _LOGV(bev_output_len(conn->upstream)),
+ "listen", _LOGV_ENDPOINT(&r->listen),
+ "upstream", _LOGV_ENDPOINT(&r->upstream)
);
+
+ drain_client_then_close(conn);
+ return;
}
free_conn(conn);
+ return;
}
- (void)bev;
+ if (events & BEV_EVENT_EOF) {
+ if (bev == conn->client && conn->close_after_client_eof) {
+ free_conn(conn);
+ return;
+ }
+
+ if (bev == conn->upstream && client_has_pending_output(conn)) {
+ drain_client_then_close(conn);
+ return;
+ }
+
+ free_conn(conn);
+ return;
+ }
}
static int connect_upstream(struct bufferevent *bev, const struct endpoint *ep)
@@ -177,13 +254,68 @@ static void pipe_read_cb(struct bufferevent *src, void *arg)
struct evbuffer *input = bufferevent_get_input(src);
struct evbuffer *output = bufferevent_get_output(dst);
+#ifdef TINYPROXY_DEBUG
+ const struct route *r = conn->route;
+
+ size_t input_len = evbuffer_get_length(input);
+ size_t output_before = evbuffer_get_length(output);
+
+ LOG_INFO("stream pipe read",
+ "line", _LOGV(r->line_no),
+ "from", _LOGV(bev_side(conn, src)),
+ "to", _LOGV(bev_side(conn, dst)),
+ "input_len", _LOGV(input_len),
+ "dst_output_before", _LOGV(output_before)
+ );
+#endif
+
evbuffer_add_buffer(output, input);
- if (evbuffer_get_length(output) >= BEV_READ_HIGH_WATER) {
+ size_t output_after = evbuffer_get_length(output);
+
+#ifdef TINYPROXY_DEBUG
+ LOG_INFO("stream pipe queued",
+ "line", _LOGV(r->line_no),
+ "from", _LOGV(bev_side(conn, src)),
+ "to", _LOGV(bev_side(conn, dst)),
+ "dst_output_after", _LOGV(output_after)
+ );
+#endif
+
+ if (output_after >= BEV_READ_HIGH_WATER) {
+#ifdef TINYPROXY_DEBUG
+ LOG_INFO("stream pipe backpressure pause",
+ "line", _LOGV(r->line_no),
+ "paused", _LOGV(bev_side(conn, src)),
+ "dst_output_len", _LOGV(output_after)
+ );
+#endif
+
bufferevent_disable(src, EV_READ);
}
}
+static void finish_client_write(conn_t *conn)
+{
+ evutil_socket_t fd = bufferevent_getfd(conn->client);
+
+ if (fd >= 0) {
+#ifndef _WIN32
+ shutdown(fd, SHUT_WR);
+#else
+ shutdown(fd, SD_SEND);
+#endif
+ }
+
+ bufferevent_disable(conn->client, EV_WRITE);
+
+ /*
+ * Keep EV_READ enabled so we can observe client EOF instead of
+ * closing with unread data and causing RST on some platforms.
+ */
+ bufferevent_enable(conn->client, EV_READ);
+}
+
static void pipe_write_cb(struct bufferevent *dst, void *arg)
{
conn_t *conn = arg;
@@ -198,13 +330,44 @@ static void pipe_write_cb(struct bufferevent *dst, void *arg)
}
struct evbuffer *output = bufferevent_get_output(dst);
+ size_t output_len = evbuffer_get_length(output);
+
+#ifdef TINYPROXY_DEBUG
+ LOG_INFO("stream pipe write",
+ "line", _LOGV(conn->route->line_no),
+ "dst", _LOGV(bev_side(conn, dst)),
+ "src", _LOGV(bev_side(conn, src)),
+ "dst_output_len", _LOGV(output_len)
+ );
+#endif
+
+ if (dst == conn->client &&
+ conn->close_client_after_drain &&
+ output_len == 0) {
+ LOG_INFO("client output drained; shutting down write side",
+ "line", _LOGV(conn->route->line_no)
+ );
+
+ finish_client_write(conn);
+ conn->close_client_after_drain = false;
+ conn->close_after_client_eof = true;
+ return;
+ }
+
+ if (src != NULL && output_len < BEV_WRITE_RESUME_WATER) {
+#ifdef TINYPROXY_DEBUG
+ LOG_INFO("stream pipe backpressure resume",
+ "line", _LOGV(conn->route->line_no),
+ "resumed", _LOGV(bev_side(conn, src)),
+ "dst", _LOGV(bev_side(conn, dst)),
+ "dst_output_len", _LOGV(output_len)
+ );
+#endif
- if (evbuffer_get_length(output) < BEV_WRITE_RESUME_WATER) {
bufferevent_enable(src, EV_READ);
}
}
-
static void worker_adopt_client_fd(struct worker *w, struct accepted_client *ac) {
conn_t *conn = calloc(1, sizeof(*conn));
if (conn == NULL) {
diff --git a/src/stream_route.h b/src/stream_route.h
index d25407b..45b38dc 100644
--- a/src/stream_route.h
+++ b/src/stream_route.h
@@ -28,6 +28,9 @@ typedef struct conn_s {
struct sockaddr_storage peer_addr;
socklen_t peer_addr_len;
+
+ bool close_after_client_eof;
+ bool close_client_after_drain;
} conn_t;
struct accepted_client {
diff --git a/tests/test_unix_listeners.py b/tests/test_unix_listeners.py
index e9427e0..c1b3060 100644
--- a/tests/test_unix_listeners.py
+++ b/tests/test_unix_listeners.py
@@ -33,6 +33,8 @@ UNIX_TO_UNIX_BACKEND_SOCK = unix_sock_path("test-pong.sock")
UNIX_DGRAM_TO_UNIX_DGRAM_LISTEN_SOCK = unix_sock_path("test-ping-dgram.sock")
UNIX_DGRAM_TO_UNIX_DGRAM_BACKEND_SOCK = unix_sock_path("test-pong-dgram.sock")
+FRONT_PORT = BACKEND_PORT + 1
+
def unlink_if_exists(path: str) -> None:
try:
os.unlink(path)
@@ -60,44 +62,6 @@ async def close_writer(writer: asyncio.StreamWriter) -> None:
except ConnectionResetError:
pass
-
-async def open_unix_connection(path: str):
- return await asyncio.open_unix_connection(path)
-
-def unix_stream_roundtrip_sync(path: str, payload: bytes) -> bytes:
- with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s:
- s.settimeout(3.0)
- s.connect(path)
- s.sendall(payload)
-
- chunks = []
- remaining = len(payload)
-
- while remaining > 0:
- data = s.recv(remaining)
- if not data:
- break
- chunks.append(data)
- remaining -= len(data)
-
- return b"".join(chunks)
-
-
-async def unix_stream_roundtrip(path: str, payload: bytes) -> bytes:
- return await asyncio.to_thread(unix_stream_roundtrip_sync, path, payload)
-
-
-def unix_stream_read_sync(path: str) -> bytes:
- with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s:
- s.settimeout(3.0)
- s.connect(path)
- return s.recv(65536)
-
-
-async def unix_stream_read(path: str) -> bytes:
- return await asyncio.to_thread(unix_stream_read_sync, path)
-
-
async def udp_echo_server(host: str, port: int):
loop = asyncio.get_running_loop()
@@ -129,77 +93,6 @@ def unix_dgram_roundtrip(sock_path: str, payload: bytes) -> bytes:
client.close()
unlink_if_exists(client_path)
-
-async def test_unix_stream_to_tcp() -> None:
- proxy_bin = os.environ.get("TINYPROXY_BIN")
- if not proxy_bin:
- raise SkipTest("TINYPROXY_BIN is not set")
-
- unlink_if_exists(UNIX_STREAM_SOCK)
-
- conf_text = (
- f"listen unix {UNIX_STREAM_SOCK} tcp {LISTEN_HOST}:{BACKEND_PORT}\n"
- )
-
- backend_server = await asyncio.start_server(
- echo_handler,
- LISTEN_HOST,
- BACKEND_PORT,
- backlog=128,
- )
-
- try:
- async with run_tinyproxy_with_conf(
- proxy_bin=proxy_bin,
- conf_text=conf_text,
- proto="unix",
- ):
- payload = b"hello over unix stream\n"
-
- got = await asyncio.wait_for(
- unix_stream_roundtrip(UNIX_STREAM_SOCK, payload),
- timeout=3.0,
- )
-
- assert got == payload, (
- f"unix stream roundtrip mismatch: got={got!r} expected={payload!r}"
- )
- finally:
- backend_server.close()
- await backend_server.wait_closed()
- unlink_if_exists(UNIX_STREAM_SOCK)
-
-
-async def test_unix_stream_to_builtin_client_addr() -> None:
- proxy_bin = os.environ.get("TINYPROXY_BIN")
- if not proxy_bin:
- raise SkipTest("TINYPROXY_BIN is not set")
-
- unlink_if_exists(UNIX_BUILTIN_SOCK)
-
- conf_text = (
- f"listen unix {UNIX_BUILTIN_SOCK} builtin client_addr\n"
- )
-
- try:
- async with run_tinyproxy_with_conf(
- proxy_bin=proxy_bin,
- conf_text=conf_text,
- proto="unix",
- ):
- got = await asyncio.wait_for(
- unix_stream_read(UNIX_BUILTIN_SOCK),
- timeout=3.0,
- )
-
- assert got, "expected builtin client_addr response"
- assert b"unix" in got.lower() or b"unknown" in got.lower() or got.strip(), (
- f"unexpected builtin client_addr response: {got!r}"
- )
- finally:
- unlink_if_exists(UNIX_BUILTIN_SOCK)
-
-
async def test_unix_dgram_to_udp() -> None:
if sys.platform.startswith("win"):
raise SkipTest("Unix socks tests are skipped on Windows")
@@ -266,22 +159,38 @@ async def test_unix_dgram_to_builtin_client_addr() -> None:
finally:
unlink_if_exists(UNIX_DGRAM_BUILTIN_SOCK)
-async def test_unix_stream_to_unix_stream() -> None:
+async def tcp_roundtrip(host: str, port: int, payload: bytes) -> bytes:
+ reader, writer = await asyncio.open_connection(host, port)
+
+ try:
+ writer.write(payload)
+ await writer.drain()
+
+ return await asyncio.wait_for(
+ reader.readexactly(len(payload)),
+ timeout=3.0,
+ )
+ finally:
+ await close_writer(writer)
+
+
+async def run_stream_chain_roundtrip(
+ conf_text: str,
+ front_port: int,
+ payload: bytes,
+ socks_to_cleanup: list[str],
+) -> bytes:
proxy_bin = os.environ.get("TINYPROXY_BIN")
if not proxy_bin:
raise SkipTest("TINYPROXY_BIN is not set")
- unlink_if_exists(UNIX_TO_UNIX_LISTEN_SOCK)
- unlink_if_exists(UNIX_TO_UNIX_BACKEND_SOCK)
-
- conf_text = (
- f"listen unix {UNIX_TO_UNIX_LISTEN_SOCK} "
- f"unix {UNIX_TO_UNIX_BACKEND_SOCK}\n"
- )
+ for path in socks_to_cleanup:
+ unlink_if_exists(path)
- backend_server = await asyncio.start_unix_server(
+ backend_server = await asyncio.start_server(
echo_handler,
- path=UNIX_TO_UNIX_BACKEND_SOCK,
+ LISTEN_HOST,
+ BACKEND_PORT,
backlog=128,
)
@@ -289,32 +198,102 @@ async def test_unix_stream_to_unix_stream() -> None:
async with run_tinyproxy_with_conf(
proxy_bin=proxy_bin,
conf_text=conf_text,
- proto="unix",
+ proto="tcp",
+ listen_port=front_port,
):
- reader, writer = await open_unix_connection(UNIX_TO_UNIX_LISTEN_SOCK)
+ return await tcp_roundtrip(LISTEN_HOST, front_port, payload)
+ finally:
+ backend_server.close()
+ await backend_server.wait_closed()
+
+ for path in socks_to_cleanup:
+ unlink_if_exists(path)
+
+async def tcp_read_once(host: str, port: int) -> bytes:
+ reader, writer = await asyncio.open_connection(host, port)
+
+ try:
+ return await asyncio.wait_for(reader.read(65536), timeout=3.0)
+ finally:
+ await close_writer(writer)
+
+async def test_unix_stream_to_unix_stream() -> None:
+ payload = b"hello unix to unix\n"
+
+ conf_text = (
+ f"listen tcp {LISTEN_HOST}:{FRONT_PORT} "
+ f"unix {UNIX_TO_UNIX_LISTEN_SOCK}\n"
+ f"listen unix {UNIX_TO_UNIX_LISTEN_SOCK} "
+ f"unix {UNIX_TO_UNIX_BACKEND_SOCK}\n"
+ f"listen unix {UNIX_TO_UNIX_BACKEND_SOCK} "
+ f"tcp {LISTEN_HOST}:{BACKEND_PORT}\n"
+ )
+
+ got = await run_stream_chain_roundtrip(
+ conf_text=conf_text,
+ front_port=FRONT_PORT,
+ payload=payload,
+ socks_to_cleanup=[
+ UNIX_TO_UNIX_LISTEN_SOCK,
+ UNIX_TO_UNIX_BACKEND_SOCK,
+ ],
+ )
+
+ assert got == payload, (
+ f"unix-to-unix roundtrip mismatch: "
+ f"got={got!r} expected={payload!r}"
+ )
+
+async def test_unix_stream_to_tcp() -> None:
+ payload = b"hello over unix stream\n"
+
+ conf_text = (
+ f"listen tcp {LISTEN_HOST}:{FRONT_PORT} "
+ f"unix {UNIX_STREAM_SOCK}\n"
+ f"listen unix {UNIX_STREAM_SOCK} "
+ f"tcp {LISTEN_HOST}:{BACKEND_PORT}\n"
+ )
+
+ got = await run_stream_chain_roundtrip(
+ conf_text=conf_text,
+ front_port=FRONT_PORT,
+ payload=payload,
+ socks_to_cleanup=[UNIX_STREAM_SOCK],
+ )
+
+ assert got == payload, (
+ f"unix stream roundtrip mismatch: got={got!r} expected={payload!r}"
+ )
+
+async def test_unix_stream_to_builtin_client_addr() -> None:
+ proxy_bin = os.environ.get("TINYPROXY_BIN")
+ if not proxy_bin:
+ raise SkipTest("TINYPROXY_BIN is not set")
- try:
- payload = b"hello unix to unix\n"
+ unlink_if_exists(UNIX_BUILTIN_SOCK)
- writer.write(payload)
- await writer.drain()
+ conf_text = (
+ f"listen tcp {LISTEN_HOST}:{FRONT_PORT} "
+ f"unix {UNIX_BUILTIN_SOCK}\n"
+ f"listen unix {UNIX_BUILTIN_SOCK} "
+ f"builtin client_addr\n"
+ )
- got = await asyncio.wait_for(
- reader.readexactly(len(payload)),
- timeout=3.0,
- )
+ try:
+ async with run_tinyproxy_with_conf(
+ proxy_bin=proxy_bin,
+ conf_text=conf_text,
+ proto="tcp",
+ listen_port=FRONT_PORT,
+ ):
+ got = await tcp_read_once(LISTEN_HOST, FRONT_PORT)
- assert got == payload, (
- f"unix-to-unix roundtrip mismatch: "
- f"got={got!r} expected={payload!r}"
- )
- finally:
- await close_writer(writer)
+ assert got, "expected builtin client_addr response"
+ assert b"unix" in got.lower() or b"unknown" in got.lower() or got.strip(), (
+ f"unexpected builtin client_addr response: {got!r}"
+ )
finally:
- backend_server.close()
- await backend_server.wait_closed()
- unlink_if_exists(UNIX_TO_UNIX_LISTEN_SOCK)
- unlink_if_exists(UNIX_TO_UNIX_BACKEND_SOCK)
+ unlink_if_exists(UNIX_BUILTIN_SOCK)
async def unix_dgram_echo_server(path: str):
loop = asyncio.get_running_loop()