From dc69a27b1dc5c9e9d0dd968c425a2bcbecd866e5 Mon Sep 17 00:00:00 2001 From: Eric Hartford Date: Sun, 18 Oct 2026 12:03:02 -0700 Subject: [PATCH] Stop rejecting slow request bodies with a silent 400 (fixes #2) Each HTTP worker set SO_RCVTIMEO to the keep-alive idle timeout (1 s) and left it on for the whole connection, so it also bounded every read inside a request. A client that paused over a second between headers and body, or mid-body, got "400 failed to read request body" with nothing on stderr, and the same request succeeded on retry. curl hits it on bodies over 1 KB: it sends Expect: 100-continue and waits a second for a 100 Continue this server never sent, racing the server's timeout. - The idle timeout now bounds only the wait for the first byte of a request; once a request starts, reads use --request-timeout-ms (default 30000), and a stall past it is a 408, not a 400. - Expect: 100-continue is answered with HTTP/1.1 100 Continue. - A reset between requests is a normal close, not an error. - Every error response is logged to stderr ("http : "). - When valid inputs fail inside the engine, the response is a 500 with type server_error instead of a 400 blaming the input; the inference service reports which case occurred. test_http_dimensions (run by make test on every platform) now checks that a body sent 1.5 s after its headers is served, that Expect gets 100 Continue followed by the response, and that a body stalled past the request timeout gets 408. Against main's binary the first check fails with the reported 400. Validated: make test and make test-gemma2 on an M5 Max (clang) and a Ryzen 9950X (GCC 15). --- AGENTS.md | 8 ++- CONTRIBUTING.md | 8 ++- src/inference_service.c | 8 ++- src/inference_service.h | 4 +- src/server.c | 106 +++++++++++++++++++++++++------ testdata/test_http_dimensions.py | 56 ++++++++++++++++ 6 files changed, 163 insertions(+), 27 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index bf5ddc7..dfabf4c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -177,8 +177,12 @@ The target is the hardware ceiling under many concurrent requests (SlimServe - **Front-end wins that cost nothing numerically:** table base64 (18x) and libcompression inflate for PNG IDAT (1.65x), both byte-identical. Compare HTTP responses with `cmp`, and send unique bodies: the response cache - serves repeats, and curl's `Expect: 100-continue` breaks this server's - body read. + serves repeats. +- **The keep-alive idle timeout used to bound the request body read** (issue + #2, fixed 2026-10-18): a client pausing over 1 s between headers and body, + or curl waiting on `Expect: 100-continue`, got a silent 400 that a retry + fixed. Reads now switch to `--request-timeout-ms` once a request starts, + `Expect` is answered, and every error response is logged. ## EmbeddingGemma 2 text lessons (2026-10-11) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index c55777b..5d6ecf5 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -655,7 +655,13 @@ Important production controls: - `--tokenizer-workers N`: set persistent tokenizer workers. - `--batch-wait-us N`: bound microbatch collection delay. - `--keepalive-connections N`, `--keepalive-max-requests N`, and - `--keepalive-timeout-ms N`: control persistent HTTP connections. + `--keepalive-timeout-ms N`: control persistent HTTP connections. The + keep-alive timeout (1 s) bounds only the wait for the next request on an + idle connection. +- `--request-timeout-ms N` (default 30000): bounds each read once a request + has started; a stalled request gets 408. `Expect: 100-continue` is answered + with `100 Continue`. Every error response is also logged to stderr, and a + failure executing valid inputs is a 500, not a 400. - `--response-cache-mb N`: size the exact float-JSON response LRU. - `--persistent-cache-path PATH`: persist the exact-embedding cache to disk. The service loads matching entries at startup and flushes ready entries on diff --git a/src/inference_service.c b/src/inference_service.c index 5cba6a4..9f229ed 100644 --- a/src/inference_service.c +++ b/src/inference_service.c @@ -1135,13 +1135,14 @@ bool ei_inference_service_embed_batch(ei_inference_service *service, size_t batch_size, float *out, char *err, size_t err_len) { return ei_inference_service_embed_batch_with_usage( - service, texts, text_lengths, batch_size, out, NULL, err, err_len); + service, texts, text_lengths, batch_size, out, NULL, NULL, err, err_len); } bool ei_inference_service_embed_batch_with_usage( ei_inference_service *service, const char *const *texts, const size_t *text_lengths, size_t batch_size, float *out, - size_t *prompt_tokens, char *err, size_t err_len) { + size_t *prompt_tokens, bool *backend_failed, char *err, size_t err_len) { + if (backend_failed) *backend_failed = false; if (!service || !service->tokenizer || !texts || !text_lengths || batch_size == 0) { set_error(err, err_len, "embedding batch must contain at least one input"); @@ -1223,7 +1224,10 @@ bool ei_inference_service_embed_batch_with_usage( if (!await_entry_locked(service, entries[i], out + i * EI_N_EMBD, item_error, sizeof item_error)) { + /* Entries fail only when the backend could not execute them; + * input validation happened before submission. */ if (ok) snprintf(first_error, sizeof first_error, "%s", item_error); + if (backend_failed) *backend_failed = true; ok = false; } } diff --git a/src/inference_service.h b/src/inference_service.h index 9508498..31fd5c1 100644 --- a/src/inference_service.h +++ b/src/inference_service.h @@ -73,10 +73,12 @@ bool ei_inference_service_embed_batch(ei_inference_service *service, const size_t *text_lengths, size_t batch_size, float *out, char *err, size_t err_len); +/* backend_failed (optional) is set when the inputs were valid but executing + * them failed (a server-side error rather than a bad request). */ bool ei_inference_service_embed_batch_with_usage( ei_inference_service *service, const char *const *texts, const size_t *text_lengths, size_t batch_size, float *out, - size_t *prompt_tokens, char *err, size_t err_len); + size_t *prompt_tokens, bool *backend_failed, char *err, size_t err_len); bool ei_inference_service_embed_tokens(ei_inference_service *service, const int32_t *ids, size_t n_tokens, float out[EI_N_EMBD], diff --git a/src/server.c b/src/server.c index fcb1284..c2a5be2 100644 --- a/src/server.c +++ b/src/server.c @@ -53,6 +53,7 @@ static int ei_socket_errno(void) { int error = WSAGetLastError(); if (error == WSAEINTR) return EINTR; if (error == WSAETIMEDOUT || error == WSAEWOULDBLOCK) return EAGAIN; + if (error == WSAECONNRESET) return ECONNRESET; return error; #else return errno; @@ -401,6 +402,7 @@ typedef struct { size_t keepalive_connections; size_t keepalive_max_requests; uint32_t keepalive_timeout_ms; + uint32_t request_timeout_ms; size_t response_cache_bytes; const char *persistent_cache_path; } server_opts; @@ -886,8 +888,15 @@ static void http_response(ei_socket fd, int status, const char *reason, body, keep_alive); } +/* One stderr line per error response, so a client-side failure report can be + * matched to the server's reason (issue #2: rejected requests left no trace). */ +static void log_http_error(int status, const char *msg) { + fprintf(stderr, "http %d: %s\n", status, msg); +} + static void http_error(ei_socket fd, int status, const char *reason, const char *msg, bool keep_alive) { + log_http_error(status, msg); sbuf b = {0}; sbuf_append_z(&b, "{\"error\":"); sbuf_append_json_string(&b, msg, strlen(msg)); @@ -901,6 +910,7 @@ static void http_error(ei_socket fd, int status, const char *reason, const char static void openai_http_error(ei_socket fd, int status, const char *reason, const char *msg, const char *param, bool keep_alive) { + log_http_error(status, msg); sbuf b = {0}; sbuf_append_z(&b, "{\"error\":{\"message\":"); sbuf_append_json_string(&b, msg, strlen(msg)); @@ -994,31 +1004,71 @@ static char *find_header_end(char *buf, size_t n) { return NULL; } +/* Two receive timeouts per connection: idle_ms bounds the wait for the next + * request on a keep-alive connection (--keepalive-timeout-ms), request_ms + * bounds each read once a request has started (--request-timeout-ms). The + * idle timeout used to govern the body read too, so a client that paused + * over a second between headers and body got a 400 (issue #2). */ typedef struct { ei_socket fd; sbuf buffered; + uint32_t idle_ms, request_ms, current_ms; } http_connection; typedef enum { HTTP_READ_OK, HTTP_READ_CLOSED, HTTP_READ_ERROR, + HTTP_READ_TIMEOUT, } http_read_result; +static void connection_timeout(http_connection *connection, uint32_t ms) { + if (connection->current_ms == ms) return; +#ifdef _WIN32 + DWORD timeout = ms; +#else + struct timeval timeout = { + .tv_sec = ms / 1000u, + .tv_usec = (suseconds_t)(ms % 1000u) * 1000, + }; +#endif + ei_setsockopt(connection->fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof timeout); + connection->current_ms = ms; +} + +static void send_all(ei_socket fd, const char *data, size_t len) { + while (len) { + int chunk = len > INT_MAX ? INT_MAX : (int)len; + ssize_t sent = send(fd, data, chunk, 0); + if (sent <= 0) { + if (sent < 0 && ei_socket_errno() == EINTR) continue; + return; + } + data += sent; + len -= (size_t)sent; + } +} + static http_read_result read_request(http_connection *connection, char **headers_out, char **body_out, size_t *body_len_out, char *err, size_t err_len) { char tmp[8192]; + connection_timeout(connection, connection->buffered.n ? connection->request_ms + : connection->idle_ms); char *hdr_end = find_header_end( connection->buffered.data, connection->buffered.n); while (!hdr_end) { ssize_t n = recv(connection->fd, tmp, sizeof tmp, 0); if (n < 0) { if (ei_socket_errno() == EINTR) continue; - if ((ei_socket_errno() == EAGAIN || ei_socket_errno() == EWOULDBLOCK) && - connection->buffered.n == 0) { + /* Between requests, a timeout or a reset is the client leaving. */ + if (connection->buffered.n == 0 && ei_socket_errno() == ECONNRESET) return HTTP_READ_CLOSED; + if (ei_socket_errno() == EAGAIN || ei_socket_errno() == EWOULDBLOCK) { + if (connection->buffered.n == 0) return HTTP_READ_CLOSED; /* idle */ + snprintf(err, err_len, "timed out reading request headers"); + return HTTP_READ_TIMEOUT; } snprintf(err, err_len, "failed to read request"); return HTTP_READ_ERROR; @@ -1030,6 +1080,7 @@ static http_read_result read_request(http_connection *connection, return HTTP_READ_ERROR; } sbuf_append(&connection->buffered, tmp, (size_t)n); + connection_timeout(connection, connection->request_ms); /* a request has started */ hdr_end = find_header_end( connection->buffered.data, connection->buffered.n); if (!hdr_end && connection->buffered.n > MAX_HEADER_BYTES) { @@ -1070,13 +1121,22 @@ static http_read_result read_request(http_connection *connection, return HTTP_READ_ERROR; } + /* A client that sent Expect: 100-continue (curl does for bodies over + * 1 KB) waits for this before sending the body. */ + if (connection->buffered.n - body_start < body_len && + header_has_token(headers, "expect", "100-continue")) { + static const char continue_line[] = "HTTP/1.1 100 Continue\r\n\r\n"; + send_all(connection->fd, continue_line, sizeof continue_line - 1u); + } while (connection->buffered.n - body_start < body_len) { ssize_t n = recv(connection->fd, tmp, sizeof tmp, 0); if (n < 0) { if (ei_socket_errno() == EINTR) continue; - snprintf(err, err_len, "failed to read request body"); + bool timed_out = ei_socket_errno() == EAGAIN || ei_socket_errno() == EWOULDBLOCK; + snprintf(err, err_len, timed_out ? "timed out reading request body" + : "failed to read request body"); free(headers); - return HTTP_READ_ERROR; + return timed_out ? HTTP_READ_TIMEOUT : HTTP_READ_ERROR; } if (n == 0) { snprintf(err, err_len, "client closed connection before body completed"); @@ -1186,7 +1246,7 @@ static void handle_embed(ei_socket fd, ei_inference_service *service, bool any_structured = false; for (size_t i = 0; i < inputs.n; ++i) any_structured |= inputs.structured[i]; - bool embedded = true; + bool embedded = true, backend_failed = false; if (any_structured) { for (size_t i = 0; i < inputs.n && embedded; ++i) { size_t count = 0; @@ -1195,13 +1255,14 @@ static void handle_embed(ei_socket fd, ei_inference_service *service, &count, err, sizeof err); } else embedded = ei_inference_service_embed_batch_with_usage( - service, texts + i, lengths + i, 1, embeddings + i * EI_N_EMBD, &count, err, - sizeof err); + service, texts + i, lengths + i, 1, embeddings + i * EI_N_EMBD, &count, + &backend_failed, err, sizeof err); prompt_tokens += count; } } else embedded = ei_inference_service_embed_batch_with_usage( - service, texts, lengths, inputs.n, embeddings, &prompt_tokens, err, sizeof err); + service, texts, lengths, inputs.n, embeddings, &prompt_tokens, &backend_failed, err, + sizeof err); if (!embedded) { free(embeddings); free(lengths); @@ -1209,8 +1270,10 @@ static void handle_embed(ei_socket fd, ei_inference_service *service, free(owned_cache_key); free_inputs(&inputs); free(model.s); - embedding_http_error(fd, api, 400, "Bad Request", err, "input", - keep_alive); + if (backend_failed) /* valid inputs the engine failed to run: not the client's fault */ + embedding_http_error(fd, api, 500, "Internal Server Error", err, NULL, keep_alive); + else + embedding_http_error(fd, api, 400, "Bad Request", err, "input", keep_alive); return; } @@ -1324,6 +1387,8 @@ static bool handle_client(http_connection *connection, if (read_result != HTTP_READ_OK) { if (read_result == HTTP_READ_ERROR) { http_error(connection->fd, 400, "Bad Request", err, false); + } else if (read_result == HTTP_READ_TIMEOUT) { + http_error(connection->fd, 408, "Request Timeout", err, false); } return false; } @@ -1687,6 +1752,7 @@ static void usage(const char *argv0) { " [--max-client-batch-size N] [--tokenizer-workers N]\n" " [--batch-wait-us N] [--keepalive-connections N]\n" " [--keepalive-max-requests N] [--keepalive-timeout-ms N]\n" + " [--request-timeout-ms N]\n" " [--response-cache-mb N] [--persistent-cache-path PATH]\n" " [--pipeline-slots N] (embeddinggemma-2 GPU waves in flight, 1..8)\n" " [--media-slots N] (embeddinggemma-2 GPU media contexts in flight, 1..8; default 2)\n" @@ -1755,6 +1821,7 @@ static bool parse_args(int argc, char **argv, server_opts *opts) { opts->keepalive_connections = 0; opts->keepalive_max_requests = 100; opts->keepalive_timeout_ms = 1000; + opts->request_timeout_ms = 30000; opts->response_cache_bytes = 64u * 1024u * 1024u; opts->persistent_cache_path = NULL; for (int i = 1; i < argc; i++) { @@ -1856,6 +1923,10 @@ static bool parse_args(int argc, char **argv, server_opts *opts) { size_t value; if (!parse_size_arg(argv[++i], 1, 60000, &value)) return false; opts->keepalive_timeout_ms = (uint32_t)value; + } else if (strcmp(argv[i], "--request-timeout-ms") == 0 && i + 1 < argc) { + size_t value; + if (!parse_size_arg(argv[++i], 1, 3600000, &value)) return false; + opts->request_timeout_ms = (uint32_t)value; } else if (strcmp(argv[i], "--response-cache-mb") == 0 && i + 1 < argc) { size_t value; if (!parse_size_arg(argv[++i], 0, 4096, &value)) return false; @@ -1963,18 +2034,11 @@ static void *server_worker_main(void *opaque) { server_worker *worker = opaque; for (;;) { ei_socket fd = socket_queue_pop(worker->queue); - http_connection connection = { .fd = fd }; -#ifdef _WIN32 - DWORD receive_timeout = worker->opts->keepalive_timeout_ms; -#else - struct timeval receive_timeout = { - .tv_sec = worker->opts->keepalive_timeout_ms / 1000u, - .tv_usec = (suseconds_t)( - worker->opts->keepalive_timeout_ms % 1000u) * 1000, + http_connection connection = { + .fd = fd, + .idle_ms = worker->opts->keepalive_timeout_ms, + .request_ms = worker->opts->request_timeout_ms, }; -#endif - ei_setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &receive_timeout, - sizeof receive_timeout); bool keepalive_slot = worker->opts->keepalive_max_requests > 1 && keepalive_limiter_try_acquire(worker->keepalive); for (size_t request = 0;; request++) { diff --git a/testdata/test_http_dimensions.py b/testdata/test_http_dimensions.py index 1565762..c1f401a 100644 --- a/testdata/test_http_dimensions.py +++ b/testdata/test_http_dimensions.py @@ -111,6 +111,59 @@ def check_keep_alive(base_url): raise AssertionError("server did not close after Connection: close") +def read_status_line(connection): + connection.settimeout(10) + data = b"" + while b"\r\n" not in data: + chunk = connection.recv(256) + if not chunk: + break + data += chunk + return data.split(b"\r\n", 1)[0].decode("ascii", "replace"), data + + +def check_request_timeouts(base_url, request_timeout_s): + """Issue #2: the keep-alive idle timeout (1 s) also bounded the body read, + so a client pausing between headers and body got a silent 400.""" + port = int(base_url.rsplit(":", 1)[1]) + body = json.dumps({"input": ["request timeout regression " + "x" * 2000]}).encode() + head = ( + "POST /api/embed HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\n" + f"Content-Length: {len(body)}\r\nConnection: close\r\n" + ).encode() + # A pause longer than the idle timeout but inside the request timeout is served. + with socket.create_connection(("127.0.0.1", port), timeout=10) as connection: + connection.sendall(head + b"\r\n") + time.sleep(1.5) + connection.sendall(body) + status, _ = read_status_line(connection) + if not status.startswith("HTTP/1.1 200 "): + raise AssertionError(f"paused body was not served: {status!r}") + # Expect: 100-continue gets an interim response, then the request is served. + with socket.create_connection(("127.0.0.1", port), timeout=10) as connection: + connection.sendall(head + b"Expect: 100-continue\r\n\r\n") + status, data = read_status_line(connection) + if status != "HTTP/1.1 100 Continue": + raise AssertionError(f"Expect: 100-continue was not answered: {status!r}") + connection.sendall(body) + rest = data.split(b"\r\n\r\n", 1)[1] if b"\r\n\r\n" in data else b"" + while b"\r\n" not in rest: + chunk = connection.recv(256) + if not chunk: + break + rest += chunk + final = rest.split(b"\r\n", 1)[0].decode("ascii", "replace") + if not final.startswith("HTTP/1.1 200 "): + raise AssertionError(f"request after 100 Continue failed: {final!r}") + # A body that stalls past the request timeout is a 408, not a 400. + with socket.create_connection(("127.0.0.1", port), timeout=10) as connection: + connection.sendall(head + b"\r\n" + body[:10]) + time.sleep(request_timeout_s + 1.0) + status, _ = read_status_line(connection) + if not status.startswith("HTTP/1.1 408 "): + raise AssertionError(f"stalled body did not time out with 408: {status!r}") + + def check_discovery(base_url): content_type, body = get(base_url + "/") if content_type != "application/json": @@ -377,6 +430,8 @@ def main(): str(port), "--workers", "4", + "--request-timeout-ms", + "2500", ] process = subprocess.Popen( command, @@ -388,6 +443,7 @@ def main(): wait_until_ready(process, base_url) check_discovery(base_url) check_keep_alive(base_url) + check_request_timeouts(base_url, 2.5) check_dimensions(base_url) check_openai_compatibility(base_url) finally: