commit 01644820644afe572e2a20baa7f3ad9c7f301e1a
| author | 斟酌 鵬兄 <tgckpg@gmail.com> |
|---|
| date | 2026-06-18T12:55:35Z |
|---|
| subject | Moved timeout logs into conn_idle_timeout_cb |
|---|
commit 01644820644afe572e2a20baa7f3ad9c7f301e1a
Author: 斟酌 鵬兄 <tgckpg@gmail.com>
Date: 2026-06-18T12:55:35Z
Moved timeout logs into conn_idle_timeout_cb
---
src/stream_conn.c | 68 +++++++++++++++-----------
tests/test_tcp_idle_timeout.py | 106 +++++++++++++++++++++++++++++++++++++++--
2 files changed, 144 insertions(+), 30 deletions(-)
diff --git a/src/stream_conn.c b/src/stream_conn.c
index 1271c1b..d705d6c 100644
--- a/src/stream_conn.c
+++ b/src/stream_conn.c
@@ -17,6 +17,8 @@
#include <event2/bufferevent.h>
#include <event2/buffer.h>
+#define SOCKADDR_STRLEN 128
+
void free_conn(conn_t *conn) {
if (conn == NULL) {
return;
@@ -70,6 +72,15 @@ static void schedule_conn_idle_timer(conn_t *conn)
evtimer_add(conn->idle_ev, &tv);
}
+static size_t bev_output_len(struct bufferevent *bev)
+{
+ if (bev == NULL) {
+ return 0;
+ }
+
+ return evbuffer_get_length(bufferevent_get_output(bev));
+}
+
static void conn_idle_timeout_cb(evutil_socket_t fd, short events, void *arg)
{
(void)events;
@@ -87,9 +98,34 @@ static void conn_idle_timeout_cb(evutil_socket_t fd, short events, void *arg)
return;
}
- LOG_DEBUG("stream connection idle timed out",
- "listen", _LOGV_ENDPOINT(&conn->route->listen),
- "upstream", _LOGV_ENDPOINT(&conn->route->upstream),
+ const struct route *r = conn->route;
+
+ char peer_buf[SOCKADDR_STRLEN];
+
+ size_t client_output = bev_output_len(conn->client);
+ size_t upstream_output = bev_output_len(conn->upstream);
+
+ if (client_output == 0 && upstream_output == 0) {
+ LOG_DEBUG("stream idle timed out",
+ "listen", _LOGV_ENDPOINT(&r->listen),
+ "upstream", _LOGV_ENDPOINT(&r->upstream),
+ "client_addr", _LOGV_SOCKADDR(&conn->peer_addr, conn->peer_addr_len,
+ peer_buf, sizeof(peer_buf)),
+ "client_sni", _LOGV(r->opts.sni_sniff ? stream_sniff_log_sni(&conn->sniff) : "")
+ );
+
+ free_conn(conn);
+ return;
+ }
+
+ LOG_WARN("stream I/O stalled timed out",
+ "client_output", _LOGV(client_output),
+ "upstream_output", _LOGV(upstream_output),
+ "listen", _LOGV_ENDPOINT(&r->listen),
+ "upstream", _LOGV_ENDPOINT(&r->upstream),
+ "client_addr", _LOGV_SOCKADDR(&conn->peer_addr, conn->peer_addr_len,
+ peer_buf, sizeof(peer_buf)),
+ "client_sni", _LOGV(r->opts.sni_sniff ? stream_sniff_log_sni(&conn->sniff) : "")
);
free_conn(conn);
@@ -132,15 +168,6 @@ static int set_socket_keepalive(evutil_socket_t fd, const struct route *r)
return 0;
}
-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;
@@ -282,7 +309,7 @@ void stream_upstream_event_cb(struct bufferevent *bev, short events, void *arg)
conn_t *conn = arg;
const struct route *r = conn->route;
- char peer_buf[128];
+ char peer_buf[SOCKADDR_STRLEN];
if (events & BEV_EVENT_CONNECTED) {
conn->upstream_connected = true;
@@ -306,20 +333,7 @@ void stream_upstream_event_cb(struct bufferevent *bev, short events, void *arg)
goto out_free;
}
- if (!client_has_pending_output(conn) && bev_output_len(conn->upstream) == 0) {
- LOG_DEBUG("stream idle timed out",
- "listen", _LOGV_ENDPOINT(&r->listen),
- "upstream", _LOGV_ENDPOINT(&r->upstream),
- "client_addr", _LOGV_SOCKADDR(&conn->peer_addr, conn->peer_addr_len,
- peer_buf, sizeof(peer_buf)),
- "client_sni", _LOGV(r->opts.sni_sniff ? stream_sniff_log_sni(&conn->sniff) : "")
- );
- goto out_free;
- }
-
- LOG_WARN("stream I/O stalled timed out",
- "client_output", _LOGV(bev_output_len(conn->client)),
- "upstream_output", _LOGV(bev_output_len(conn->upstream)),
+ LOG_ERROR("unreachable code reached at BEV_EVENT_TIMEOUT",
"listen", _LOGV_ENDPOINT(&r->listen),
"upstream", _LOGV_ENDPOINT(&r->upstream),
"client_addr", _LOGV_SOCKADDR(&conn->peer_addr, conn->peer_addr_len,
diff --git a/tests/test_tcp_idle_timeout.py b/tests/test_tcp_idle_timeout.py
index d875270..12c121e 100644
--- a/tests/test_tcp_idle_timeout.py
+++ b/tests/test_tcp_idle_timeout.py
@@ -58,6 +58,29 @@ async def delayed_response_upload_handler(
writer.close()
await writer.wait_closed()
+async def stalled_upload_handler(
+ reader: asyncio.StreamReader,
+ writer: asyncio.StreamWriter,
+) -> None:
+ try:
+ try:
+ await reader.readuntil(b"\r\n\r\n")
+ except asyncio.IncompleteReadError as e:
+ # Ignore readiness/preflight connections from the test harness.
+ if not e.partial:
+ return
+ raise AssertionError(f"incomplete request headers: {e.partial!r}") from e
+
+ # Read a little bit so the request is definitely established,
+ # then stop consuming the upload body. This should eventually
+ # backpressure tinyproxy and trigger the stalled I/O timeout.
+ await reader.readexactly(64 * 1024)
+
+ await asyncio.sleep(10.0)
+ finally:
+ writer.close()
+ await writer.wait_closed()
+
async def echo_handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
try:
while True:
@@ -71,15 +94,13 @@ async def echo_handler(reader: asyncio.StreamReader, writer: asyncio.StreamWrite
writer.close()
await writer.wait_closed()
-
async def close_writer(writer: asyncio.StreamWriter) -> None:
try:
writer.close()
await writer.wait_closed()
- except ConnectionResetError:
+ except OSError:
pass
-
async def read_one_or_eof(reader: asyncio.StreamReader, timeout: float = 3.0) -> bytes:
try:
return await asyncio.wait_for(reader.read(1), timeout=timeout)
@@ -259,8 +280,87 @@ async def test_tcp_idle_timeout_keeps_long_one_way_upload_open() -> None:
finally:
await backend_server.close()
+async def test_tcp_stalled_timeout_should_close() -> None:
+ proxy_bin = os.environ.get("TINYPROXY_BIN")
+ if not proxy_bin:
+ raise SkipTest("TINYPROXY_BIN is not set")
+
+ conf_text = (
+ f"listen"
+ f" tcp {LISTEN_HOST}:{PROXY_PORT}"
+ f" tcp {LISTEN_HOST}:{BACKEND_PORT}"
+ f" idle_timeout=2\n"
+ )
+
+ backend_server = await start_tracked_stream_server(
+ stalled_upload_handler,
+ LISTEN_HOST,
+ BACKEND_PORT,
+ backlog=128,
+ )
+
+ try:
+ async with run_tinyproxy_with_conf(
+ proxy_bin=proxy_bin,
+ conf_text=conf_text,
+ listen_host=LISTEN_HOST,
+ listen_port=PROXY_PORT,
+ proto="tcp",
+ ):
+ reader, writer = await asyncio.open_connection(LISTEN_HOST, PROXY_PORT)
+
+ try:
+ body_size = 512 * 1024 * 1024
+ chunk_size = 1024 * 1024
+
+ req = (
+ b"PUT /v2/test/blobs/uploads/stalled?digest=sha256:deadbeef HTTP/1.1\r\n"
+ b"Host: fake-registry\r\n"
+ b"User-Agent: test-buildkit/stalled-repro\r\n"
+ b"Content-Type: application/octet-stream\r\n"
+ + f"Content-Length: {body_size}\r\n".encode("ascii")
+ + b"Connection: close\r\n"
+ + b"\r\n"
+ )
+
+ writer.write(req)
+ await writer.drain()
+
+ chunk = b"x" * chunk_size
+ sent = 0
+ closed = False
+
+ deadline = asyncio.get_running_loop().time() + 8.0
+
+ while asyncio.get_running_loop().time() < deadline:
+ try:
+ writer.write(chunk)
+ await asyncio.wait_for(writer.drain(), timeout=3.0)
+ sent += len(chunk)
+ except (
+ ConnectionResetError,
+ BrokenPipeError,
+ asyncio.TimeoutError,
+ ):
+ closed = True
+ break
+
+ if not closed:
+ got = await read_one_or_eof(reader, timeout=1.0)
+ closed = got == b""
+
+ assert closed, (
+ f"expected stalled upload to be closed by idle timeout; "
+ f"sent={sent}"
+ )
+ finally:
+ await close_writer(writer)
+ finally:
+ await backend_server.close()
+
TESTS = [
("test_tcp_idle_timeout_closes_idle_connection", test_tcp_idle_timeout_closes_idle_connection),
("test_tcp_idle_timeout_keeps_active_connection_open", test_tcp_idle_timeout_keeps_active_connection_open),
("test_tcp_idle_timeout_keeps_long_one_way_upload_open", test_tcp_idle_timeout_keeps_long_one_way_upload_open),
+ ("test_tcp_stalled_timeout_should_close", test_tcp_stalled_timeout_should_close),
]