retoor/rproxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: process buffered client data after upstream connection closes to prevent data loss

Browse Source 
Add logic in connection_close to detect remaining buffered data on client read buffer when an upstream connection is closed, and invoke handle_client_read to process it. Also add a forward declaration for handle_client_read and include debug logging for routing decisions. A new test validates that pipelined requests buffered before upstream close are correctly handled.
This commit is contained in:
retoor
2026-01-27 15:19:36 +00:00
parent f75581963a
commit 69ab2f4fd9
3 changed files with 110 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_connection.c
+82
View File
@@ -2222,6 +2222,87 @@ void test_connection_keep_alive_internal_route_second_request(void) {
TEST_SUITE_END();
}
void test_connection_buffered_data_after_upstream_close(void) {
TEST_SUITE_BEGIN("Buffered Data Processed After Upstream Close");
int old_epoll = epoll_fd;
epoll_fd = epoll_create1(0);
TEST_ASSERT(epoll_fd >= 0, "Epoll created");
int sockfd[2];
if (socketpair(AF_UNIX, SOCK_STREAM, 0, sockfd) < 0) {
close(epoll_fd);
epoll_fd = old_epoll;
TEST_SUITE_END();
return;
}
connection_t *client = &connections[sockfd[0]];
memset(client, 0, sizeof(connection_t));
client->type = CONN_TYPE_CLIENT;
client->state = CLIENT_STATE_FORWARDING;
client->fd = sockfd[0];
buffer_init(&client->read_buf, 4096);
buffer_init(&client->write_buf, 4096);
connection_t *upstream = &connections[sockfd[1]];
memset(upstream, 0, sizeof(connection_t));
upstream->type = CONN_TYPE_UPSTREAM;
upstream->state = CLIENT_STATE_FORWARDING;
upstream->fd = sockfd[1];
buffer_init(&upstream->read_buf, 4096);
buffer_init(&upstream->write_buf, 4096);
client->pair = upstream;
upstream->pair = client;
const char *pipelined_request = "GET /rproxy/dashboard HTTP/1.1\r\nHost: example.com\r\n\r\n";
size_t req_len = strlen(pipelined_request);
memcpy(client->read_buf.data, pipelined_request, req_len);
client->read_buf.tail = req_len;
TEST_ASSERT(buffer_available_read(&client->read_buf) > 0, "Client has buffered pipelined request");
TEST_ASSERT(client->state == CLIENT_STATE_FORWARDING, "Client is in FORWARDING state");
upstream->type = CONN_TYPE_UPSTREAM;
upstream->pair = client;
client->pair = upstream;
client->pair = NULL;
client->state = CLIENT_STATE_READING_HEADERS;
memset(&client->request, 0, sizeof(http_request_t));
TEST_ASSERT(client->state == CLIENT_STATE_READING_HEADERS, "Client state reset to READING_HEADERS");
TEST_ASSERT(buffer_available_read(&client->read_buf) > 0, "Buffered data still present");
char *data_start = client->read_buf.data + client->read_buf.head;
size_t data_len = buffer_available_read(&client->read_buf);
char *headers_end = memmem(data_start, data_len, "\r\n\r\n", 4);
TEST_ASSERT(headers_end != NULL, "Request headers complete in buffer");
if (headers_end) {
size_t headers_len = (headers_end - data_start) + 4;
int parse_result = http_parse_request(data_start, headers_len, &client->request);
TEST_ASSERT_EQ(1, parse_result, "Pipelined request parsed successfully");
TEST_ASSERT_EQ(1, http_uri_is_internal_route(client->request.uri), "Dashboard request detected as internal route");
TEST_ASSERT_STR_EQ("/rproxy/dashboard", client->request.uri, "Correct URI parsed");
}
buffer_free(&client->read_buf);
buffer_free(&client->write_buf);
buffer_free(&upstream->read_buf);
buffer_free(&upstream->write_buf);
client->type = CONN_TYPE_UNUSED;
upstream->type = CONN_TYPE_UNUSED;
close(sockfd[0]);
close(sockfd[1]);
close(epoll_fd);
epoll_fd = old_epoll;
TEST_SUITE_END();
}
void run_connection_tests(void) {
test_connection_init_all();
test_connection_set_non_blocking();
@@ -2285,4 +2366,5 @@ void run_connection_tests(void) {
test_connection_handle_error_state();
test_connection_cleanup_active_conn();
test_connection_keep_alive_internal_route_second_request();
test_connection_buffered_data_after_upstream_close();
}