diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/8408-fetch-connection-reuse.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/8408-fetch-connection-reuse.ts.json new file mode 100644 index 000000000..0a8304063 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/8408-fetch-connection-reuse.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/8408-fetch-connection-reuse.ts" + ], + "diags": [] +} diff --git a/packages/runtime/src/scr_fetch.c b/packages/runtime/src/scr_fetch.c index 1b30c451d..652dfa8f1 100644 --- a/packages/runtime/src/scr_fetch.c +++ b/packages/runtime/src/scr_fetch.c @@ -5386,9 +5386,21 @@ static void sf_on_data(ScrClosure *cb, ScrBytes *chunk) { sf_release(t); } +/* After 'end' the keep-alive agent can hand the socket to the next request + * on the connection, so the transfer leaves it reading and stops pausing + * and resuming it. */ +static void sf_drop_response_socket(SfTransfer *t) { + if (!t->response_socket) return; + if (t->response_paused) scr_net_sock_release(scr_net_sock_resume(t->response_socket)); + t->response_paused = false; + scr_net_sock_release(t->response_socket); + t->response_socket = NULL; +} + static void sf_on_end(ScrClosure *cb) { SfTransfer *t = sf_from(cb); if (!t) return; + sf_drop_response_socket(t); if (!t->done) { if (t->inflating && t->response_stream && !t->response_stream->discarded) { @@ -5929,9 +5941,16 @@ static bool sf_start_hop(SfTransfer *t) { wrap_ctx = scr_tls_fetch_client_ctx(sni, true); wrap = &scr_tls_fetch_client_wrap; } - ScrHttpClientReq *client = scr_http_request_ex( - dial, dial_port, path, t->method, 0, headers, false, NULL, NULL, - default_port, wrap, wrap_ctx); + /* A direct connection goes through fetch's keep-alive agent, which hands + * an idle connection to the next request for the same origin, as + * undici's dispatcher does; the bare host keys the pool, so origins that + * share an address keep their own connections. A proxied request dials + * its own connection. */ + ScrHttpClientReq *client = + proxy ? scr_http_request_ex(dial, dial_port, path, t->method, 0, headers, false, NULL, + NULL, default_port, wrap, wrap_ctx) + : scr_http_fetch_request(https, dial, dial_port, path, t->method, headers, + default_port, wrap, wrap_ctx, bare); scr_arr_release(headers); scr_str_release(path); scr_str_release(dial); diff --git a/packages/runtime/src/scr_http.c b/packages/runtime/src/scr_http.c index 0284c31c4..b006d9c36 100644 --- a/packages/runtime/src/scr_http.c +++ b/packages/runtime/src/scr_http.c @@ -5286,6 +5286,9 @@ typedef struct ScrHttpAgent { double max_sockets; /* INFINITY = Node's default */ double max_free; double timeout_ms; /* < 0 = unset */ + /* without a timeout, how long an idle pooled socket stays open (0 = until + * the server closes it); fetch's agent keeps undici's 4 s */ + double idle_ms; double default_port; /* settable (agent.defaultPort = p) */ bool destroyed; ScrHttpAgentEnt *ents; /* append order — Node's FIFO queue */ @@ -5519,7 +5522,7 @@ static bool scr_http_client_pool(struct ScrHttpConn *conn, struct ScrHttpClientR scr_str_release(name); return false; } - double timeout = ag->timeout_ms > 0 ? ag->timeout_ms : 0; + double timeout = ag->timeout_ms > 0 ? ag->timeout_ms : ag->idle_ms; if (hint >= 0) { double server = hint - 1000; if (server < 0) { @@ -5606,20 +5609,12 @@ static ScrStr *scr_http_agent_name(const char *host, size_t host_len, const char return scr_jb_finish(&b); } -ScrDyn *scr_http_agent_new(bool secure, bool keep_alive, double ka_msecs, - double max_sockets, double max_free, double timeout_ms, - double port /* < 0 = unset */, const ScrDyn *scheduling) { +/* A registered agent (+1), the half of new http.Agent(opts) after option + * validation. */ +static ScrHttpAgent *scr_http_agent_alloc(bool secure, bool keep_alive, double ka_msecs, + double max_sockets, double max_free, double timeout_ms, + double port, bool fifo) { scr_http_install(); - /* Node: options.scheduling || 'lifo', then validateOneOf */ - bool fifo = false; - if (scheduling != NULL && scr_dyn_truthy(scheduling)) { - const ScrStr *s = scheduling->kind == SCR_DYN_STR ? scheduling->v.str : NULL; - if (s != NULL && s->len == 4 && memcmp(s->data, "fifo", 4) == 0) fifo = true; - else if (s == NULL || s->len != 4 || memcmp(s->data, "lifo", 4) != 0) { - scr_dyn_arg_value_fail("scheduling", "must be one of: 'fifo', 'lifo'", scheduling); - return NULL; - } - } ScrHttpAgent *a = calloc(1, sizeof *a); if (!a) scr_http_oom(); a->rc = 1; @@ -5642,6 +5637,25 @@ ScrDyn *scr_http_agent_new(bool secure, bool keep_alive, double ka_msecs, a->in_registry = true; a->next = scr_http_agents; scr_http_agents = scr_http_agent_retain(a); + return a; +} + +ScrDyn *scr_http_agent_new(bool secure, bool keep_alive, double ka_msecs, + double max_sockets, double max_free, double timeout_ms, + double port /* < 0 = unset */, const ScrDyn *scheduling) { + scr_http_install(); + /* Node: options.scheduling || 'lifo', then validateOneOf */ + bool fifo = false; + if (scheduling != NULL && scr_dyn_truthy(scheduling)) { + const ScrStr *s = scheduling->kind == SCR_DYN_STR ? scheduling->v.str : NULL; + if (s != NULL && s->len == 4 && memcmp(s->data, "fifo", 4) == 0) fifo = true; + else if (s == NULL || s->len != 4 || memcmp(s->data, "lifo", 4) != 0) { + scr_dyn_arg_value_fail("scheduling", "must be one of: 'fifo', 'lifo'", scheduling); + return NULL; + } + } + ScrHttpAgent *a = scr_http_agent_alloc(secure, keep_alive, ka_msecs, max_sockets, max_free, + timeout_ms, port, fifo); ScrDyn *d = scr_dyn_new_handle(a, SCR_DYNH_HTTP_AGENT); scr_http_agent_release(a); /* the handle's retain holds it */ return d; @@ -5703,6 +5717,59 @@ static SCR_TL ScrStr *scr_http_next_socket_path = NULL; extern SCR_TL ScrNetSocket *(*scr_http_socket_dialer)(ScrStr *path, bool deferred); SCR_TL ScrNetSocket *(*scr_http_socket_dialer)(ScrStr *path, bool deferred) = NULL; +/* A request through agent `ag`: an idle pooled socket for the request's + * name (and pool_key) serves it, a free slot dials, and otherwise it + * queues. socket_path (borrowed, nullable) dials a Unix socket. */ +static ScrHttpClientReq *scr_http_request_pooled(ScrHttpAgent *ag, ScrStr *host /*borrowed*/, + double port, ScrStr *path /*borrowed*/, + ScrStr *method /*borrowed*/, double timeout_ms, + ScrArr *header_pairs /*borrowed*/, bool auto_end, + ScrClosure *cb /*moves, nullable*/, + ScrHttpRespFn fn, int default_port, + void (*wrap)(ScrNetSocket *, void *), + void *wrap_ctx, ScrStr *pool_key /*borrowed*/, + ScrStr *socket_path /*borrowed*/) { + double p = port >= 0 ? port + : ag->default_port > 0 ? ag->default_port + : (double)default_port; + char portbuf[16]; + int portn = snprintf(portbuf, sizeof portbuf, "%d", (int)p); + ScrStr *name = scr_http_agent_name(host->data, host->len, portbuf, (size_t)portn, + NULL, 0, 0, socket_path ? socket_path->data : NULL, + socket_path ? socket_path->len : 0); + /* a free pooled socket for the name serves the request right away */ + ScrHttpAgentFree *reuse = ag->keep_alive ? scr_http_agent_free_take(ag, name, pool_key) : NULL; + bool queue = reuse == NULL && scr_http_agent_active(ag, name) >= ag->max_sockets; + ScrNetSocket *presock = socket_path != NULL ? scr_http_socket_dialer(socket_path, queue || reuse) + : queue || reuse ? scr_net_connect_deferred(p, host) + : NULL; + ScrHttpClientReq *c = scr_http_request_impl(host, p, path, method, timeout_ms, header_pairs, + auto_end, cb, fn, default_port, wrap, wrap_ctx, + presock); + if (c == NULL) { /* the method-token throw: nothing registered */ + if (reuse != NULL) scr_http_agent_free_untake(ag, reuse); /* back to the list it came from */ + scr_str_release(name); + return NULL; + } + c->pool_key = pool_key ? scr_str_retain(pool_key) : NULL; + if (ag->timeout_ms >= 0 && timeout_ms <= 0) { + c->timeout_ms = ag->timeout_ms; + scr_net_sock_set_timeout(c->sock, ag->timeout_ms); + } + if (reuse != NULL) scr_http_client_rebind(c, reuse, true); + else if (!queue) scr_http_conn_count(c->conn, ag->sock_count); /* dialed now */ + c->agent = scr_http_agent_retain(ag); + ScrHttpAgentEnt *e = calloc(1, sizeof *e); + if (!e) scr_http_oom(); + e->name = name; /* ownership moves */ + e->client = scr_http_client_retain(c); + e->queued = queue; + ScrHttpAgentEnt **link = &ag->ents; + while (*link) link = &(*link)->next; + *link = e; + return c; +} + ScrHttpClientReq *scr_http_request_agent_ex(ScrStr *host /*borrowed*/, double port, ScrStr *path /*borrowed*/, ScrStr *method /*borrowed*/, double timeout_ms, ScrArr *header_pairs /*borrowed*/, @@ -5754,46 +5821,29 @@ ScrHttpClientReq *scr_http_request_agent_ex(ScrStr *host /*borrowed*/, double po scr_dyn_arg_type_fail("options.agent", "an instance of http.Agent, false, null, or undefined", agent); return NULL; } - ScrHttpAgent *ag = (ScrHttpAgent *)agent->v.handle.ptr; - double p = port >= 0 ? port - : ag->default_port > 0 ? ag->default_port - : (double)default_port; - char portbuf[16]; - int portn = snprintf(portbuf, sizeof portbuf, "%d", (int)p); - ScrStr *name = scr_http_agent_name(host->data, host->len, portbuf, (size_t)portn, - NULL, 0, 0, socket_path ? socket_path->data : NULL, - socket_path ? socket_path->len : 0); - /* a free pooled socket for the name serves the request right away */ - ScrHttpAgentFree *reuse = ag->keep_alive ? scr_http_agent_free_take(ag, name, pool_key) : NULL; - bool queue = reuse == NULL && scr_http_agent_active(ag, name) >= ag->max_sockets; - ScrNetSocket *presock = socket_path != NULL ? scr_http_socket_dialer(socket_path, queue || reuse) - : queue || reuse ? scr_net_connect_deferred(p, host) - : NULL; - ScrHttpClientReq *c = scr_http_request_impl(host, p, path, method, timeout_ms, header_pairs, - auto_end, cb, fn, default_port, wrap, wrap_ctx, - presock); - if (c == NULL) { /* the method-token throw: nothing registered */ - if (reuse != NULL) scr_http_agent_free_untake(ag, reuse); /* back to the list it came from */ - scr_str_release(name); - return NULL; + return scr_http_request_pooled((ScrHttpAgent *)agent->v.handle.ptr, host, port, path, method, + timeout_ms, header_pairs, auto_end, cb, fn, default_port, wrap, + wrap_ctx, pool_key, socket_path); +} + +/* The agents fetch() dispatches through, like undici's global dispatcher: + * keep-alive connections per origin with no socket limit and no timeout on + * a request in flight; an idle connection closes after 4 s, or the server's + * Keep-Alive hint less one second when that is shorter. */ +static SCR_TL ScrHttpAgent *scr_http_fetch_agents[2]; + +ScrHttpClientReq *scr_http_fetch_request(bool secure, ScrStr *host /*borrowed*/, double port, + ScrStr *path /*borrowed*/, ScrStr *method /*borrowed*/, + ScrArr *header_pairs /*borrowed*/, int default_port, + void (*wrap)(ScrNetSocket *, void *), void *wrap_ctx, + ScrStr *pool_key /*borrowed*/) { + ScrHttpAgent **slot = &scr_http_fetch_agents[secure ? 1 : 0]; + if (*slot == NULL) { + *slot = scr_http_agent_alloc(secure, true, -1, -1, -1, -1, -1, false); + (*slot)->idle_ms = 4000; } - c->pool_key = pool_key ? scr_str_retain(pool_key) : NULL; - if (ag->timeout_ms >= 0 && timeout_ms <= 0) { - c->timeout_ms = ag->timeout_ms; - scr_net_sock_set_timeout(c->sock, ag->timeout_ms); - } - if (reuse != NULL) scr_http_client_rebind(c, reuse, true); - else if (!queue) scr_http_conn_count(c->conn, ag->sock_count); /* dialed now */ - c->agent = scr_http_agent_retain(ag); - ScrHttpAgentEnt *e = calloc(1, sizeof *e); - if (!e) scr_http_oom(); - e->name = name; /* ownership moves */ - e->client = scr_http_client_retain(c); - e->queued = queue; - ScrHttpAgentEnt **link = &ag->ents; - while (*link) link = &(*link)->next; - *link = e; - return c; + return scr_http_request_pooled(*slot, host, port, path, method, 0, header_pairs, false, NULL, NULL, + default_port, wrap, wrap_ctx, pool_key, NULL); } ScrHttpClientReq *scr_http_request_agent(ScrStr *host /*borrowed*/, double port, @@ -5811,6 +5861,8 @@ static void scr_http_agents_cleanup(void) { for (size_t i = 0; i < 2; i++) { scr_dyn_release(scr_http_global_agents[i]); scr_http_global_agents[i] = NULL; + if (scr_http_fetch_agents[i]) scr_http_agent_release(scr_http_fetch_agents[i]); + scr_http_fetch_agents[i] = NULL; } while (scr_http_agents) { ScrHttpAgent *a = scr_http_agents; diff --git a/packages/runtime/src/scr_http.h b/packages/runtime/src/scr_http.h index 654362c06..c07f12c6a 100644 --- a/packages/runtime/src/scr_http.h +++ b/packages/runtime/src/scr_http.h @@ -365,6 +365,14 @@ ScrHttpClientReq *scr_http_request_agent_ex(ScrStr *host /*borrowed*/, double po ScrStr *pool_key /*borrowed, nullable: https TLS options*/); /* +1 */ /* http.globalAgent / https.globalAgent: borrowed, and as a +1 value. */ const ScrDyn *scr_http_global_agent(bool secure); +/* fetch()'s client request for a direct connection: through fetch's own + * keep-alive agent, which reuses an idle connection with the same host, + * port and pool_key (+1). */ +ScrHttpClientReq *scr_http_fetch_request(bool secure, ScrStr *host /*borrowed*/, double port, + ScrStr *path /*borrowed*/, ScrStr *method /*borrowed*/, + ScrArr *header_pairs /*borrowed*/, int default_port, + void (*wrap)(ScrNetSocket *, void *), void *wrap_ctx, + ScrStr *pool_key /*borrowed*/); ScrDyn *scr_http_global_agent_value(bool secure); ScrHttpClientReq *scr_http_request_agent(ScrStr *host /*borrowed*/, double port, ScrStr *path /*borrowed*/, ScrStr *method /*borrowed*/, diff --git a/tests/corpus/8408-fetch-connection-reuse.ts b/tests/corpus/8408-fetch-connection-reuse.ts new file mode 100644 index 000000000..f36946629 --- /dev/null +++ b/tests/corpus/8408-fetch-connection-reuse.ts @@ -0,0 +1,95 @@ +// fetch() reuses keep-alive connections as Node's fetch does: sequential +// requests share connections, and every response is the same with reuse: +// concurrent requests, `connection: close` responses and requests, +// redirects, request bodies, an aborted response followed by another +// request, and a request after the server closed its idle connections. The +// program exits while idle connections remain open. +import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; +import type { AddressInfo } from "node:net"; + +let connections = 0; +const server = createServer((req: IncomingMessage, res: ServerResponse) => { + const url = req.url ?? "/"; + if (url === "/close") { + res.writeHead(200, { connection: "close" }); + res.end("closing"); + return; + } + if (url === "/redirect") { + res.writeHead(302, { location: "/hello" }); + res.end(); + return; + } + if (url === "/echo") { + let body = ""; + req.setEncoding("utf8"); + req.on("data", (chunk: string) => { + body += chunk; + }); + req.on("end", () => res.end(`echo:${body}`)); + return; + } + if (url === "/slow") { + res.writeHead(200, { "content-type": "text/plain" }); + res.write("first "); + setTimeout(() => res.end("rest"), 200); + return; + } + res.end(`hello ${url}`); +}); +server.on("connection", () => { + connections++; +}); + +async function text(port: number, path: string, init?: RequestInit): Promise { + const res = await fetch(`http://127.0.0.1:${port}${path}`, init); + return `${res.status} ${await res.text()}`; +} + +async function main(port: number): Promise { + const seen: string[] = []; + for (let i = 0; i < 5; i++) seen.push(await text(port, `/seq${i}`)); + console.log("sequential", connections < 5, seen.join(" | ")); + + const parallel = await Promise.all([1, 2, 3].map((i) => text(port, `/par${i}`))); + for (let i = 0; i < 3; i++) parallel.push(await text(port, `/after${i}`)); + console.log("concurrent", parallel.join(" | ")); + + console.log(await text(port, "/close"), await text(port, "/close")); + console.log(await text(port, "/a", { headers: { connection: "close" } }), await text(port, "/b")); + + const res = await fetch(`http://127.0.0.1:${port}/redirect`); + console.log("redirect", res.redirected, res.url.endsWith("/hello"), await res.text()); + + const bodies = []; + for (let i = 0; i < 3; i++) bodies.push(await text(port, "/echo", { method: "POST", body: `n${i}` })); + console.log("post", bodies.join(" | ")); + + const controller = new AbortController(); + const slow = await fetch(`http://127.0.0.1:${port}/slow`, { signal: controller.signal }); + controller.abort(); + try { + await slow.text(); + } catch (e) { + console.log("aborted", (e as Error).name); + } + console.log(await text(port, "/after-abort")); + + // A server that closes idle connections after 50 ms: the next request + // after a pause dials again. + const brief = createServer((req: IncomingMessage, res: ServerResponse) => res.end(`brief ${req.url}`)); + brief.keepAliveTimeout = 50; + await new Promise((resolve) => brief.listen(0, "127.0.0.1", () => resolve())); + const briefPort = (brief.address() as AddressInfo).port; + console.log(await text(briefPort, "/1"), await text(briefPort, "/2")); + await new Promise((resolve) => setTimeout(resolve, 200)); + console.log(await text(briefPort, "/3")); + + brief.close(); + server.close(); + console.log("done"); +} + +server.listen(0, "127.0.0.1", () => { + void main((server.address() as AddressInfo).port); +}); diff --git a/tests/harness/corpus-costs/8408-fetch-connection-reuse.ts.json b/tests/harness/corpus-costs/8408-fetch-connection-reuse.ts.json new file mode 100644 index 000000000..a5a948601 --- /dev/null +++ b/tests/harness/corpus-costs/8408-fetch-connection-reuse.ts.json @@ -0,0 +1 @@ +{ "plain": 9400, "san": 30000 }