From 82cc41ab0a42608ced9e76ed2ba36892b7eaba8a Mon Sep 17 00:00:00 2001 From: Chris Tate Date: Thu, 23 Jul 2026 22:40:41 -0500 Subject: [PATCH] Checkpoint: dyn-handle compat members and the http Agent surface - ServerResponse/IncomingMessage/Socket members that fenced on checked-dynamic values now dispatch: flushHeaders, getHeaders, setTimeout, cork/uncork with a real writableCorked counter and coalescing flush, writeHead's raw-array form, write/end encodings and deferred callbacks, req.pause/resume with buffered drain, destroyed/readable/bytesWritten/res.req reads, setNoDelay, destroySoon, end(cb), and dyn Buffer at()/slice() - new http.Agent/https.Agent lowers to a checked-dynamic handle: literal option parsing, getName, destroy, sockets/requests/freeSockets snapshots, settable defaultPort, and REAL maxSockets accounting (over-limit requests defer their dial and queue); keepAlive: true throws the named pooling fence - the request path threads the agent (http.requestAgent/https.requestAgent rows, both backends); dyn keys now stringify for indexing and the in-operator over dyn receivers (dyn.hasKey) - differential fixtures 2573-2577 pin both families against Node --- .../src/backend/emission/emit-exprs.ts | 43 + packages/compiler/src/backend/llvm/emitter.ts | 11 +- .../src/frontend/lowering/lower-calls.ts | 3 + .../src/frontend/lowering/lower-classes.ts | 11 +- .../src/frontend/lowering/lower-exprs.ts | 22 +- .../src/frontend/lowering/lower-server.ts | 175 ++- .../src/frontend/lowering/surfaces.ts | 10 + packages/compiler/src/ir/nodes.ts | 34 + packages/compiler/src/ir/validate.ts | 16 +- packages/runtime/src/scr_dyn_invoke.c | 30 +- packages/runtime/src/scr_http.c | 1106 ++++++++++++++++- packages/runtime/src/scr_http2.c | 1 + packages/runtime/src/scr_json.c | 21 + packages/runtime/src/scr_net.c | 257 +++- packages/runtime/src/scr_runtime.h | 70 ++ packages/runtime/src/scr_tls.c | 12 + tests/corpus/2573-http-dyn-res-compat.cjs | 87 ++ tests/corpus/2574-http-dyn-req-pause.cjs | 51 + tests/corpus/2575-net-dyn-socket-compat.cjs | 53 + tests/corpus/2576-http-agent-surface.cjs | 44 + tests/corpus/2577-http-agent-maxsockets.cjs | 38 + tests/diagnostics/options-record.ts | 7 +- 22 files changed, 2015 insertions(+), 87 deletions(-) create mode 100644 tests/corpus/2573-http-dyn-res-compat.cjs create mode 100644 tests/corpus/2574-http-dyn-req-pause.cjs create mode 100644 tests/corpus/2575-net-dyn-socket-compat.cjs create mode 100644 tests/corpus/2576-http-agent-surface.cjs create mode 100644 tests/corpus/2577-http-agent-maxsockets.cjs diff --git a/packages/compiler/src/backend/emission/emit-exprs.ts b/packages/compiler/src/backend/emission/emit-exprs.ts index 411cb2b99..1dacb95f1 100644 --- a/packages/compiler/src/backend/emission/emit-exprs.ts +++ b/packages/compiler/src/backend/emission/emit-exprs.ts @@ -2183,6 +2183,10 @@ export function emitExpr(E: CEmitter, e: IrExpr): Temp { // Object.defineProperties over DOM values: both borrowed, // result the target (+1); throws catchably (may-throw seed). return finish(`scr_dyn_define_props(${arg(0)}, ${arg(1)})`); + case "dyn.hasKey": + // `k in v` with a runtime key: the DOM presence answer (both + // borrowed, no allocation, never throws). + return finish(`scr_dyn_has_key(${arg(0)}, ${arg(1)})`); case "dyn.keySet": // Keyed write on a dyn receiver: all three borrowed (the // member retains the value in); throws Node's TypeErrors on @@ -3636,6 +3640,28 @@ export function emitExpr(E: CEmitter, e: IrExpr): Temp { `scr_http_request(${arg(0)}, ${arg(1)}, ${arg(2)}, ${arg(3)}, ${arg(4)}, ${arg(5)}, ${arg(6)}, ${cbExpr}, ${adapter})`, ); } + case "http.agentNew": + E.usesTimers = true; // queued dials hold the loop open + return finish( + `scr_http_agent_new(${arg(0)}, ${arg(1)}, ${arg(2)}, ${arg(3)}, ${arg(4)}, ${arg(5)}, ${arg(6)})`, + ); + case "http.requestAgent": + case "http.requestAgentCb": { + E.usesTimers = true; + let cbExpr = "NULL"; + let adapter = "NULL"; + if (e.fn === "http.requestAgentCb") { + const cbT = e.args[8]!.type; + if (cbT.kind !== "func") throw new Error("emitter bug: http.requestAgentCb callback not a func"); + const cb = args[8]!; + E.moveTemp(cb); + cbExpr = cb.name; + adapter = cbT.params.length === 0 ? "&scr_http_resp_thunk0" : "&scr_http_resp_thunk_res"; + } + return finish( + `scr_http_request_agent(${arg(0)}, ${arg(1)}, ${arg(2)}, ${arg(3)}, ${arg(4)}, ${arg(5)}, ${arg(6)}, ${arg(7)}, ${cbExpr}, ${adapter})`, + ); + } case "http.requestUrl": case "http.requestUrlCb": { E.usesTimers = true; // an in-flight request holds the loop open @@ -4152,6 +4178,23 @@ export function emitExpr(E: CEmitter, e: IrExpr): Temp { `scr_https_request(${arg(0)}, ${arg(1)}, ${arg(2)}, ${arg(3)}, ${arg(4)}, ${arg(5)}, ${arg(6)}, ${arg(7)}, (const char *)${arg(8)}->data, ${arg(8)}->len, ${cbExpr}, ${adapter})`, ); } + case "https.requestAgent": + case "https.requestAgentCb": { + E.usesTimers = true; // an in-flight request holds the loop open + let cbExpr = "NULL"; + let adapter = "NULL"; + if (e.fn === "https.requestAgentCb") { + const cbT = e.args[10]!.type; + if (cbT.kind !== "func") throw new Error("emitter bug: https.requestAgentCb callback not a func"); + const cb = args[10]!; + E.moveTemp(cb); + cbExpr = cb.name; + adapter = cbT.params.length === 0 ? "&scr_http_resp_thunk0" : "&scr_http_resp_thunk_res"; + } + return finish( + `scr_https_request_agent(${arg(0)}, ${arg(1)}, ${arg(2)}, ${arg(3)}, ${arg(4)}, ${arg(5)}, ${arg(6)}, ${arg(7)}, (const char *)${arg(8)}->data, ${arg(8)}->len, ${arg(9)}, ${cbExpr}, ${adapter})`, + ); + } case "https.requestFn": case "https.requestFnCb": { // The requestFn binding's runtime dial: arg 0 picks the client diff --git a/packages/compiler/src/backend/llvm/emitter.ts b/packages/compiler/src/backend/llvm/emitter.ts index bab93f194..9f7cdfc6e 100644 --- a/packages/compiler/src/backend/llvm/emitter.ts +++ b/packages/compiler/src/backend/llvm/emitter.ts @@ -457,6 +457,7 @@ const LIB_FN_SYMS: Record = { // ambient-this read never throw. The fs dyn read is the sync-fs story. "json.parse": "scr_json_parse", "dyn.keySet": "scr_dyn_key_set", + "dyn.hasKey": "scr_dyn_has_key", "dyn.defineProps": "scr_dyn_define_props", "dyn.typeof": "scr_dyn_typeof", "dyn.toString": "scr_dyn_to_string", @@ -595,6 +596,7 @@ const LIB_FN_SYMS: Record = { "http.clientDestroy": "scr_http_client_destroy", "http.clientDestroyed": "scr_http_client_destroyed", "http2.streamUndefCall": "scr_http2_stream_undef_call", + "http.agentNew": "scr_http_agent_new", // The island surface (--dynamic): eval/import bridge catchably (the // may-throw seed's pending check); importDyn answers an engine promise // and never throws itself. insp.jsval throws on composite island @@ -718,6 +720,7 @@ const USES_TIMERS_LIB_FNS = new Set([ "http.createServer", "http.createServerEmpty", "http.request", "http.requestCb", "http.requestUrl", "http.requestUrlCb", "http.requestConn", "http.requestConnCb", + "http.agentNew", "http.requestAgent", "http.requestAgentCb", // The dyn-async slice (emit-exprs.ts's markings): fiber parks, the // microtask/immediate mints, the tracing-promise reaction fiber, and // the loop-end unhandled-rejection report. @@ -10824,9 +10827,11 @@ class LlEmitter { B.line(`call void @${entry}(ptr ${args[0]!.name}, ptr ${args[1]!.name}, ptr @${adapter}, i1 ${args[2]!.name})`); return { name: "", type: e.type }; } - if (e.fn === "http.request" || e.fn === "http.requestCb" || e.fn === "http.requestUrl" || e.fn === "http.requestUrlCb") { + if (e.fn === "http.request" || e.fn === "http.requestCb" || e.fn === "http.requestUrl" || e.fn === "http.requestUrlCb" || + e.fn === "http.requestAgent" || e.fn === "http.requestAgentCb") { const isUrl = e.fn.startsWith("http.requestUrl"); - const cbIdx = isUrl ? 3 : 7; + const isAgent = e.fn.startsWith("http.requestAgent"); + const cbIdx = isUrl ? 3 : isAgent ? 8 : 7; const hasCb = e.fn.endsWith("Cb"); const args = e.args.map((a) => this.emitExpr(a)); let cb = "null"; @@ -10842,7 +10847,7 @@ class LlEmitter { } const head = args.slice(0, cbIdx); const decls = head.map((a) => (this.llType(a.type) === "i1" ? "i1 zeroext" : this.llType(a.type))); - const entry = isUrl ? "scr_http_request_url" : "scr_http_request"; + const entry = isUrl ? "scr_http_request_url" : isAgent ? "scr_http_request_agent" : "scr_http_request"; this.declare(`declare ptr @${entry}(${[...decls, "ptr", "ptr"].join(", ")})`); const t = B.tmp(); B.line( diff --git a/packages/compiler/src/frontend/lowering/lower-calls.ts b/packages/compiler/src/frontend/lowering/lower-calls.ts index 201fe2ad8..ea9ebcd22 100644 --- a/packages/compiler/src/frontend/lowering/lower-calls.ts +++ b/packages/compiler/src/frontend/lowering/lower-calls.ts @@ -3945,6 +3945,9 @@ export const DYN_DISPATCH_METHODS = new Set([ "writeContinue", "writeEarlyHints", "cork", "uncork", "addTrailers", "ref", "unref", "address", "setNoDelay", "setKeepAlive", "connect", "resetAndDestroy", "destroySoon", + // The Agent handle's own member (no other DOM prototype declares it, + // so the remainder keeps the stored-member answers). + "getName", // The netServer half of the handle surface (`let server; server = // createServer(...)` — the handle lives in a dyn binding whose // closures the checker cannot narrow): listen/close dispatch onto the diff --git a/packages/compiler/src/frontend/lowering/lower-classes.ts b/packages/compiler/src/frontend/lowering/lower-classes.ts index df5475731..3fef99ffb 100644 --- a/packages/compiler/src/frontend/lowering/lower-classes.ts +++ b/packages/compiler/src/frontend/lowering/lower-classes.ts @@ -16,7 +16,7 @@ import { requiresDynamicPackageDiag, unsupportedDiag } from "../../diagnostics/d import { STREAM_API_MEMBERS, STREAM_PROP_MEMBERS, UNDERSCORE_METHODS, lowerStreamNew, lowerStreamSuperCall, streamCtorShape } from "./lower-stream.js"; import { declSymbolOf } from "./lower-modules.js"; import { uniqueSymbolKeyOf } from "./lower-exprs.js"; -import { lowerHttpServerNew } from "./lower-server.js"; +import { lowerHttpAgentNew, lowerHttpServerNew } from "./lower-server.js"; import { ambientNsRootOf, ambientUndefReadType, ambientUndefVarRootOf, ambientUndefinedFnSymbolOf, fenceEarlyAliasUse, fenceEarlyNsMemberRef, nsMemberIdentOf, nsUndefRead } from "./lower-namespaces.js"; import { mixinResultBindingClassOf, type MixinInstanceInfo } from "./lower-mixins.js"; @@ -4806,6 +4806,13 @@ export function lowerNew(L: Lowerer, expr: ts.NewExpression): IrExpr { const httpServer = lowerHttpServerNew(L, expr); if (httpServer) return httpServer; } + // `new http.Agent(opts?)` / `new https.Agent(opts?)` — the Agent + // handle (lower-server): getName/destroy/counters through the DOM + // handle ops, requests thread it via the agent option. + { + const agent = lowerHttpAgentNew(L, expr); + if (agent) return agent; + } if (ts.isIdentifier(expr.expression)) { // `import C = N.C; new C()` — the alias's own source-order guards // (a no-op for every non-import= binding). @@ -5509,7 +5516,7 @@ export function lowerNew(L: Lowerer, expr: ts.NewExpression): IrExpr { { const STDLIB_CTOR_HINTS: Record = { Agent: - "constructing an http Agent (compiled clients dial one connection per request and close it with the response — pass agent: false for Node's one-shot semantics, or omit the option)", + "constructing an http Agent through an indirect class binding (spell the construction on the module binding — new http.Agent(...)/new https.Agent(...) or the named Agent import — which lowers to the Agent handle)", }; const ctorName = ts.isIdentifier(expr.expression) ? expr.expression diff --git a/packages/compiler/src/frontend/lowering/lower-exprs.ts b/packages/compiler/src/frontend/lowering/lower-exprs.ts index a31548339..b34055f10 100644 --- a/packages/compiler/src/frontend/lowering/lower-exprs.ts +++ b/packages/compiler/src/frontend/lowering/lower-exprs.ts @@ -5749,7 +5749,10 @@ export function lowerObjectLiteral(L: Lowerer, expr: ts.ObjectLiteralExpression) litKey !== null ? ({ kind: "strLit", value: litKey, type: STRING, loc: locOf(keyNode) } satisfies IrExpr) : L.lowerExpr(keyNode); - if (dk.type.kind === "f64") dk = L.ensureString(dk, keyNode); + // Number AND checked-dynamic keys ride the JS-exact formatter — + // property keys ARE strings (o[k] is o[String(k)] in JS), and a dyn + // key (agent.sockets[agent.getName(...)]) stringifies the same way. + if (dk.type.kind === "f64" || dk.type.kind === "dyn") dk = L.ensureString(dk, keyNode); if (dk.type.kind !== "string") { L.unsupported("SC1090", keyNode, "indexing records with non-string or non-number keys"); } @@ -8300,6 +8303,23 @@ export function lowerBinary(L: Lowerer, expr: ts.BinaryExpression): IrExpr { // runtime key table to ask. const rIn = lowerRuntimeKeyIn(L, expr, loc); if (rIn) return rIn; + // A runtime key over a CHECKED-DYNAMIC receiver (`name in + // agent.sockets` — both sides computed): the DOM presence answer, + // with the key stringified like every property key (o[k] is + // o[String(k)] in JS; `in` shares the coercion). + { + const recvT = L.mapTypeOf(L.typeOf(expr.right)); + if (recvT?.kind === "dyn") { + let k = L.lowerExpr(expr.left); + if (k.type.kind === "f64" || k.type.kind === "string" || k.type.kind === "dyn") { + k = L.ensureString(k, expr.left); + const recvD = L.lowerExpr(expr.right); + if (recvD.type.kind === "dyn") { + return { kind: "libCall", fn: "dyn.hasKey", args: [recvD, k], type: BOOL, loc }; + } + } + } + } L.unsupported( "SC1090", expr.left, diff --git a/packages/compiler/src/frontend/lowering/lower-server.ts b/packages/compiler/src/frontend/lowering/lower-server.ts index 432f856f4..3267370c1 100644 --- a/packages/compiler/src/frontend/lowering/lower-server.ts +++ b/packages/compiler/src/frontend/lowering/lower-server.ts @@ -13,6 +13,7 @@ import type { Lowerer } from "./lowerer.js"; import { locOf } from "../program.js"; import { arrayOf, BOOL, BYTES_U8, canBoxFuncIntoDyn, canConvertToDyn, DYN, DYN_HANDLE_KINDS, F64, funcOf, HTTP2SESSION_T, HTTP2STREAM_T, HTTPCLIENTREQ_T, HTTPREQ_T, HTTPRES_T, IrExpr, IrLibFn, IrStmt, IrType, NETSERVER_T, NETSOCKET_T, NULL_T, SECURECTX_T, STRING, UNDEFINED_T, SrcLoc, typeKey, VOID } from "../../ir/nodes.js"; import { + AGENT_DOCUMENTED_OPTIONS, builtinFenceHintOf, builtinModuleFnOf, fenceOrDropOptionKey, @@ -2128,6 +2129,126 @@ function lowerHttpCreateServerForms(L: Lowerer, expr: ts.CallExpression | ts.New * createServer (Node's Server class IS the factory's product); called * from lowerNew ahead of the stdlib-constructor fence. Null when the * callee is not the http module's Server. */ +/** `new http.Agent(opts?)` / `new https.Agent(opts?)` (property access or + * named import): the Agent lowers to a checked-dynamic HANDLE — + * getName/destroy and the sockets/requests/freeSockets counters dispatch + * through the DOM handle ops, and the request path threads it (the + * requestAgent rows). Options parse at the LITERAL construction site: + * keepAlive (a boolean value — TRUE throws the runtime's named pooling + * fence: this client dials one connection per request), keepAliveMsecs / + * maxSockets / maxFreeSockets / timeout / port as numbers, scheduling + * DROPS (no free pool exists, so selection order cannot observe), other + * documented keys fence by name, and undocumented keys drop like Node + * drops them. Null when the callee isn't the http/https Agent. */ +export function lowerHttpAgentNew(L: Lowerer, expr: ts.NewExpression): IrExpr | null { + const callee = expr.expression; + const bi = + ts.isPropertyAccessExpression(callee) && callee.name.text === "Agent" + ? L.builtinMemberOf(callee) + : ts.isIdentifier(callee) + ? L.builtinImportOf(callee) + : null; + if (!bi || bi.member !== "Agent" || (bi.module !== "http" && bi.module !== "https")) return null; + const loc = locOf(expr); + const api = `new ${bi.module}.Agent`; + const numLit = (value: number): IrExpr => ({ kind: "numLit", value, type: F64, loc }); + const boolAt = (value: boolean): IrExpr => ({ kind: "boolLit", value, type: BOOL, loc }); + let keepAlive: IrExpr = boolAt(false); + let kaMsecs: IrExpr = numLit(-1); + let maxSockets: IrExpr = numLit(-1); + let maxFree: IrExpr = numLit(-1); + let timeout: IrExpr = numLit(-1); + let port: IrExpr = numLit(-1); + const args = expr.arguments ?? []; + if (args.length > 1) { + L.noLowering(`${api} with ${args.length} arguments`, expr, "the supported form is new Agent(options?)"); + } + if (args.length === 1) { + const optsNode = stripParensAndCasts(args[0]!); + if (!ts.isObjectLiteralExpression(optsNode)) { + L.noLowering( + `${api} with a non-literal options value`, + args[0]!, + "spell the options as an object literal at the construction site", + ); + } + for (const prop of optsNode.properties) { + if (ts.isSpreadAssignment(prop)) { + L.noLowering(`${api} with an options spread`, prop, "write each option inline"); + } + let initializer: ts.Expression | null; + if (ts.isPropertyAssignment(prop) && + (ts.isIdentifier(prop.name) || ts.isStringLiteral(prop.name))) { + initializer = prop.initializer; + } else if (ts.isShorthandPropertyAssignment(prop)) { + initializer = null; + } else { + L.noLowering( + `${api} options with computed keys`, + prop, + "each option must be a plain `name: value` (or shorthand) entry with a literal key", + ); + } + const key = (prop.name as ts.Identifier | ts.StringLiteral).text; + const lowerVal = (want: "bool" | "f64"): IrExpr => { + const v = initializer !== null + ? L.lowerExpr(initializer) + : L.lowerShorthandValue(prop as ts.ShorthandPropertyAssignment); + if (v.type.kind === "dyn") { + return { kind: "dynCheck", value: v, type: want === "bool" ? BOOL : F64, loc: locOf(prop) }; + } + if (v.type.kind !== want) { + L.noLowering( + `a ${api} '${key}' option of '${L.fmt(v.type)}' values`, + prop, + want === "bool" ? "the option value must be a boolean" : "the option value must be a number", + ); + } + return v; + }; + switch (key) { + case "keepAlive": + keepAlive = lowerVal("bool"); + break; + case "keepAliveMsecs": + kaMsecs = lowerVal("f64"); + break; + case "maxSockets": + maxSockets = lowerVal("f64"); + break; + case "maxFreeSockets": + maxFree = lowerVal("f64"); + break; + case "timeout": + timeout = lowerVal("f64"); + break; + case "port": + // Node merges agent options under portless request options — + // the settable defaultPort carries this one. + port = lowerVal("f64"); + break; + case "scheduling": + // Free-socket selection order: no free pool exists here (no + // keep-alive reuse), so the choice cannot observe — dropped. + if (initializer !== null) L.lowerExpr(initializer); // evaluate, like Node + break; + default: + fenceOrDropOptionKey( + L, prop, key, api, AGENT_DOCUMENTED_OPTIONS, + "keepAlive, keepAliveMsecs, maxSockets, maxFreeSockets, timeout, port, and scheduling are the supported options", + ); + } + } + } + return { + kind: "libCall", + fn: "http.agentNew", + args: [boolAt(bi.module === "https"), keepAlive, kaMsecs, maxSockets, maxFree, timeout, port], + type: DYN, + loc, + }; +} + export function lowerHttpServerNew(L: Lowerer, expr: ts.NewExpression): IrExpr | null { const callee = expr.expression; const isHttpServer = @@ -3120,6 +3241,8 @@ function lowerHttpClientCall(L: Lowerer, expr: ts.CallExpression, member: "reque let connCb: IrExpr | null = null; /** agent: false — inject Connection: close into the request head. */ let agentClose: ts.Node | null = null; + /** agent: — the requestAgent rows thread the handle. */ + let agentVal: IrExpr | null = null; for (const prop of optsNode.properties) { // Shorthand entries ({ port } for { port: port }) are the natural // spelling at these call sites — the shorthand's VALUE lowering @@ -3295,8 +3418,11 @@ function lowerHttpClientCall(L: Lowerer, expr: ts.CallExpression, member: "reque // model, so it LOWERS (the header injects below). null/undefined // pick the default agent, whose keep-alive request header the // compiled client already sends (socket REUSE stays the - // documented divergence either way). Agent instances configure - // pooling the runtime does not have — fence. + // documented divergence either way). An Agent VALUE (the lowered + // `new http.Agent(...)` handle — a checked-dynamic value) threads + // through the requestAgent rows: getName-keyed maxSockets + // accounting over one-dial-per-request connections; a runtime + // non-Agent value is Node's ERR_INVALID_ARG_TYPE. const e = initializer !== null ? stripParensAndCasts(initializer) : null; if (e !== null && e.kind === ts.SyntaxKind.FalseKeyword) { agentClose = prop; @@ -3307,13 +3433,18 @@ function lowerHttpClientCall(L: Lowerer, expr: ts.CallExpression, member: "reque (ts.isIdentifier(e) && e.text === "undefined"))) { break; // the default agent: what the agent-free call compiles } - L.noLowering( - `a ${member} 'agent' option carrying an Agent value`, - prop, - "compiled clients dial one connection per request and close it with the response — " + - "agent: false (lowered: sends Connection: close) and agent: null/undefined (the default) " + - "compile; pooling Agent instances have no lowering", - ); + const v = initializer !== null + ? L.lowerExpr(initializer) + : L.lowerShorthandValue(prop as ts.ShorthandPropertyAssignment); + if (v.type.kind !== "dyn") { + L.noLowering( + `a ${member} 'agent' option of '${L.fmt(v.type)}' values`, + prop, + "the agent is the checked-dynamic Agent handle new http.Agent(...) answers " + + "(or false / null / undefined)", + ); + } + agentVal = v; break; } default: @@ -3363,11 +3494,28 @@ function lowerHttpClientCall(L: Lowerer, expr: ts.CallExpression, member: "reque "the dialer supplies the socket — set the Host header via headers: { host: ... } instead", ); } + if (agentVal !== null && connCb !== null) { + L.noLowering( + `${member} mixing an agent value with createConnection`, + optsNode, + "both own the dial — pass one or the other", + ); + } + if (agentVal !== null && binding !== null) { + L.noLowering( + `${member} through a module-function binding with an agent value`, + optsNode, + "call http.request/https.request directly when passing an Agent", + ); + } host ??= strLit("localhost"); // The binding mode's default port follows the runtime dial: 443 on the // TLS arm, 80 on the plain one — exactly each client's own default. + // With an AGENT the sentinel -1 says "no port option": the runtime + // consults the agent's (settable) defaultPort first, Node's merge. port ??= binding !== null ? { kind: "ternary", cond: secureExpr(), then: numLit(443), else_: numLit(80), type: F64, loc } + : agentVal !== null ? numLit(-1) : numLit(secure === true ? 443 : 80); path ??= strLit("/"); method ??= strLit("GET"); @@ -3393,9 +3541,12 @@ function lowerHttpClientCall(L: Lowerer, expr: ts.CallExpression, member: "reque ca ??= strLit(""); /* none: /etc/ssl/cert.pem stands in for Node's roots */ base.push(reject, ca); } + if (agentVal !== null) base.push(agentVal); if (binding !== null) base.unshift(secureExpr()); if (args.length === 1) { - const fn: IrLibFn = binding !== null ? "https.requestFn" : secure === true ? "https.request" : "http.request"; + const fn: IrLibFn = binding !== null ? "https.requestFn" + : secure === true ? (agentVal !== null ? "https.requestAgent" : "https.request") + : agentVal !== null ? "http.requestAgent" : "http.request"; return { kind: "libCall", fn, args: base, type: HTTPCLIENTREQ_T, loc }; } const { cb } = lowerCallbackArg( @@ -3404,7 +3555,9 @@ function lowerHttpClientCall(L: Lowerer, expr: ts.CallExpression, member: "reque "use (res) or ()", [HTTPREQ_T], ); - const fn: IrLibFn = binding !== null ? "https.requestFnCb" : secure === true ? "https.requestCb" : "http.requestCb"; + const fn: IrLibFn = binding !== null ? "https.requestFnCb" + : secure === true ? (agentVal !== null ? "https.requestAgentCb" : "https.requestCb") + : agentVal !== null ? "http.requestAgentCb" : "http.requestCb"; return { kind: "libCall", fn, args: [...base, cb], type: HTTPCLIENTREQ_T, loc }; } diff --git a/packages/compiler/src/frontend/lowering/surfaces.ts b/packages/compiler/src/frontend/lowering/surfaces.ts index 7ece457d9..e006e8ce2 100644 --- a/packages/compiler/src/frontend/lowering/surfaces.ts +++ b/packages/compiler/src/frontend/lowering/surfaces.ts @@ -172,6 +172,16 @@ export const HTTP_SERVER_DOCUMENTED_OPTIONS: ReadonlySet = new Set([ "requireHostHeader", "uniqueHeaders", ]); +/** new http.Agent(options)'s documented keys (Node v24 — the Agent + * constructor table plus the socket.connect() options it forwards into + * each dial). The lowered set is keepAlive/keepAliveMsecs/maxSockets/ + * maxFreeSockets/timeout/port (+ scheduling, dropped: no free pool + * exists); the rest fence by name and unknown keys drop like Node. */ +export const AGENT_DOCUMENTED_OPTIONS: ReadonlySet = new Set([ + "family", "hints", "host", "keepAliveInitialDelay", "localAddress", + "localPort", "lookup", "maxTotalSockets", "noDelay", "path", +]); + /** https.request adds tls.connect's client-side TLS knobs. */ export const HTTPS_CLIENT_DOCUMENTED_OPTIONS: ReadonlySet = new Set([ ...HTTP_CLIENT_DOCUMENTED_OPTIONS, diff --git a/packages/compiler/src/ir/nodes.ts b/packages/compiler/src/ir/nodes.ts index bff018717..a176a7993 100644 --- a/packages/compiler/src/ir/nodes.ts +++ b/packages/compiler/src/ir/nodes.ts @@ -1512,6 +1512,11 @@ export type IrLibFn = * SEMANTICS.md notes the sloppy divergence: loud, never silent). Void * result; in the may-throw seed set. */ | "dyn.keySet" + /** `key in v` with a RUNTIME (string) key on a checked-dynamic + * receiver (args: value dyn, key string; result bool): OBJ answers + * own-member presence, ARR answers 'length'/a valid index — exactly + * the compile-time dynHasKey fold, per value. Never throws. */ + | "dyn.hasKey" /** Object.defineProperties over DOM values (args: target, descriptors — * both borrowed dyn; result: the target, +1 — JS's return value). * Value descriptors become plain own properties on OBJ and FUNC targets @@ -2234,6 +2239,27 @@ export type IrLibFn = * Content-Length exactly like Node. */ | "http.request" | "http.requestCb" + /** new http.Agent(opts) / new https.Agent(opts): (secure, keepAlive, + * keepAliveMsecs, maxSockets, maxFreeSockets, timeoutMs, port) — the + * numeric options arrive < 0 for "unset" (Infinity/256/none; port + * seeds the settable defaultPort, Node's option merge). Returns the + * Agent as a checked-dynamic HANDLE (getName/destroy and the + * sockets/requests/freeSockets counters dispatch through the DOM + * handle ops). keepAlive: true THROWS the named construction fence — + * socket POOLING is not modeled (one dial per request); maxSockets + * accounting is real: over-limit requests defer their dial and queue. + * MAY THROW. */ + | "http.agentNew" + /** The agent-threaded request rows: the http.request/https.request + * shape with a trailing `agent` dyn argument (an Agent handle, false — + * the one-shot Connection: close dial — or null/undefined for the + * default path). port < 0 means "no port option": the agent's settable + * defaultPort, then the scheme's. MAY THROW (a non-Agent value is + * Node's ERR_INVALID_ARG_TYPE). */ + | "http.requestAgent" + | "http.requestAgentCb" + | "https.requestAgent" + | "https.requestAgentCb" /** The createConnection form (the proxy's own dialer): args are * (connCb, path, method, timeout, headers, autoEnd[, cb]) — connCb is * a `() => net.Socket` closure the runtime invokes ONCE, synchronously @@ -6070,6 +6096,14 @@ export const MAY_THROW_LIB_FNS: ReadonlySet = new Set([ "http.requestCb", "http.requestConn", "http.requestConnCb", + // The agent rows add the runtime agent-value gates (a non-Agent value + // is Node's ERR_INVALID_ARG_TYPE); agentNew's keepAlive: true throws + // the named pooling fence. + "http.agentNew", + "http.requestAgent", + "http.requestAgentCb", + "https.requestAgent", + "https.requestAgentCb", "https.request", "https.requestCb", "https.requestFn", diff --git a/packages/compiler/src/ir/validate.ts b/packages/compiler/src/ir/validate.ts index 766739aed..f059646c6 100644 --- a/packages/compiler/src/ir/validate.ts +++ b/packages/compiler/src/ir/validate.ts @@ -91,6 +91,7 @@ export const LIB_FN_SIGS: Recordkind == SCR_DYN_BYTES && dyn_bytes_proto_real(method)) { - dyn_throw_unsupported("Uint8Array", method); - return NULL; + if (recv->kind == SCR_DYN_BYTES) { + ScrBytes *bytes = recv->v.bytes; + size_t blen = bytes->len; + if (dyn_name_is(method, "at")) { + double iD = dyn_index_arg(args, argc, 0, 0, what); + if (scr_exc_pending()) return NULL; + double idx = iD < 0 ? (double)blen + iD : iD; + if (idx < 0 || idx >= (double)blen) return scr_dyn_retain(scr_dyn_undefined()); + return scr_dyn_new_num((double)bytes->data[(size_t)idx]); + } + if (dyn_name_is(method, "slice") || dyn_name_is(method, "subarray")) { + /* Both COPY (no views in this runtime — the static lane's + * documented divergence for subarray/Buffer.slice); the result + * keeps the receiver's Buffer flavor. */ + double startD = dyn_index_arg(args, argc, 0, 0, what); + if (scr_exc_pending()) return NULL; + double endD = dyn_index_arg(args, argc, 1, (double)blen, what); + if (scr_exc_pending()) return NULL; + ScrBytes *out = scr_bytes_slice(bytes, startD, endD); + ScrDyn *d = recv->buffer ? scr_dyn_new_buffer_copy(out) : scr_dyn_new_bytes_copy(out); + scr_bytes_release(out); + return d; + } + if (dyn_bytes_proto_real(method)) { + dyn_throw_unsupported("Uint8Array", method); + return NULL; + } } /* NUM/BOOL/BYTES-remainder: the name is no method of this kind — JS's diff --git a/packages/runtime/src/scr_http.c b/packages/runtime/src/scr_http.c index 1b428ea4b..a773d3289 100644 --- a/packages/runtime/src/scr_http.c +++ b/packages/runtime/src/scr_http.c @@ -49,6 +49,7 @@ #include "scr_runtime.h" #include +#include /* INFINITY — the Agent's maxSockets default */ #include #include #include @@ -148,6 +149,16 @@ struct ScrHttpReq { bool close_queued; bool close_emitted; /* settled: err/close listeners dropped */ bool join_dup; /* joinDuplicateHeaders: repeated names read joined ", " */ + /* pause(): delivery holds — arrived body bytes buffer in pend (the + * parser keeps consuming; memory is the buffer, a documented bound) + * and 'end' defers (end_pending) until resume() drains through the + * emit queue, never the resuming stack. */ + bool paused; + bool end_pending; + bool drain_queued; + char *pend; + size_t pend_len, pend_cap; + bool destroyed; /* req.destroy()/the teardown ran — req.destroyed */ ScrNetLs aborted_ls; /* req.on('aborted') — the h2 compat event */ }; @@ -211,6 +222,7 @@ void scr_http_req_release(ScrHttpReq *r) { scr_http_res_release(r->pipe_res); scr_http_client_release(r->pipe_client); scr_net_sock_release(r->pipe_sock); + free(r->pend); if (r->sock) scr_net_sock_release(r->sock); if (r->h2_stream) scr_http_h2_ops->release(r->h2_stream); #ifdef SCR_RC_AUDIT @@ -414,14 +426,53 @@ ScrNetSocket *scr_http_req_socket(ScrHttpReq *r) { return r->sock ? scr_net_sock_retain(r->sock) : NULL; } -/* resume(): a flow-control no-op — this parser always consumes body - * bytes and fires 'end' regardless of consumers (SEMANTICS.md; Node - * requires the stream to flow). */ -void scr_http_req_resume(ScrHttpReq *r) { (void)r; } +/* the deferred-emit queue lives below (the proto sweep) — resume()'s + * drain rides it; the parser's deliver and the cork flush live below + * their callers too */ +static void scr_http_emit_push(int kind, void *h /*moves +1*/); +static void scr_http_req_deliver(ScrHttpReq *r, const char *data, size_t n); +static void scr_http_res_cork_flush(ScrHttpRes *r); +#define SCR_HTTP_EMIT_REQ_DRAIN_K 6 + +/* resume(): a flow-control no-op unless pause() held delivery — then the + * buffered bytes (and a deferred 'end') drain through the emit queue, + * never the resuming stack. This parser always consumes body bytes + * (SEMANTICS.md; Node requires the stream to flow). */ +void scr_http_req_resume(ScrHttpReq *r) { + if (!r->paused) return; + r->paused = false; + if ((r->pend_len > 0 || r->end_pending) && !r->drain_queued) { + r->drain_queued = true; + scr_http_emit_push(SCR_HTTP_EMIT_REQ_DRAIN_K, scr_http_req_retain(r)); + } +} + +/* pause(): delivery holds until resume() (scr_http_req_deliver buffers; + * a completed body's 'end' waits too). */ +void scr_http_req_pause(ScrHttpReq *r) { + if (!r->ended) r->paused = true; +} + +/* req.setTimeout(ms[, cb]): the underlying socket's idle timer — the cb + * registers once('timeout') there, Node's delegation. Finished/destroyed + * messages skip (Node no-ops once the stream is done). */ +void scr_http_req_set_timeout(ScrHttpReq *r, double ms, ScrClosure *cb /*moves, nullable*/) { + if (r->ended || r->destroyed || r->close_emitted || !r->sock) { + /* the message is done — Node no-ops there (never arms, never fires) */ + if (cb) scr_closure_release(cb); + return; + } + scr_net_sock_set_timeout(r->sock, ms); + if (cb) scr_net_sock_on_timeout(r->sock, cb, true); +} + +bool scr_http_req_destroyed_flag(ScrHttpReq *r) { return r->destroyed; } +bool scr_http_req_readable(ScrHttpReq *r) { return !r->ended && !r->destroyed; } /* destroy(): tears the underlying connection down NOW; the teardown * events flow through the socket's native hooks like a peer close. */ void scr_http_req_destroy(ScrHttpReq *r) { + r->destroyed = true; if (r->h2_stream != NULL) { scr_http_h2_ops->destroy(r->h2_stream); return; @@ -460,6 +511,11 @@ void scr_http_req_on_aborted(ScrHttpReq *r, ScrClosure *cb /*moves*/, bool once) * req cannot cycle past the body). */ static void scr_http_req_finish(ScrHttpReq *r, bool fire) { if (r->ended) return; + if (fire && r->paused) { + /* pause() holds 'end' too — resume()'s drain finishes the body */ + r->end_pending = true; + return; + } r->ended = true; if (fire) scr_net_fire0_this(&r->end_ls, r, SCR_DYNH_HTTP_REQ); scr_net_ls_drop(&r->data_ls); @@ -503,9 +559,24 @@ struct ScrHttpRes { bool finished; bool no_date; /* res.sendDate = false: suppress the implicit Date header */ bool keep_alive; /* the REQUEST's verdict; Connection: close overrides */ + bool destroyed; /* res.destroy()/teardown — res.destroyed (true in 'close') */ + /* cork()/uncork(): corked counts the nesting (res.writableCorked); + * writes while corked coalesce in cork_buf and flush as ONE write when + * the count reaches zero (or at end()) — Node's coalescing, and the + * h2 lane's single DATA frame. */ + int corked; + char *cork_buf; + size_t cork_len, cork_cap; + /* res.req: the paired request (+1; req never points back — no cycle). + * A `res.req = null` write clears the READ (req_cleared) — Node's + * plain-property write. */ + ScrHttpReq *req_ref; + bool req_cleared; ScrNetLs close_ls; ScrNetLs finish_ls; /* res.end(cb) — fires deferred once the body went out */ + ScrNetLs wcb_ls; /* res.write(chunk, cb) — fires from the queue */ bool finish_queued; + bool wcb_queued; bool close_queued; bool close_emitted; }; @@ -527,6 +598,9 @@ void scr_http_res_release(ScrHttpRes *r) { scr_str_release(r->status_msg); scr_net_ls_drop(&r->close_ls); scr_net_ls_drop(&r->finish_ls); + scr_net_ls_drop(&r->wcb_ls); + free(r->cork_buf); + if (r->req_ref) scr_http_req_release(r->req_ref); if (r->sock) scr_net_sock_release(r->sock); if (r->h2_stream) scr_http_h2_ops->release(r->h2_stream); #ifdef SCR_RC_AUDIT @@ -761,8 +835,28 @@ static void scr_http_res_send_head(ScrHttpRes *r, long long body_len) { free(b.data); } +/* Writes while corked coalesce here and flush as ONE write when the + * cork count reaches zero (or at end()) — Node's coalescing, and one + * DATA frame on the h2 lane. */ +static void scr_http_res_cork_buffer(ScrHttpRes *r, const char *data, size_t len) { + if (len == 0) return; + if (r->cork_len + len > r->cork_cap) { + size_t cap = r->cork_cap ? r->cork_cap : 1024; + while (cap < r->cork_len + len) cap *= 2; + r->cork_buf = realloc(r->cork_buf, cap); + if (!r->cork_buf) scr_http_oom(); + r->cork_cap = cap; + } + memcpy(r->cork_buf + r->cork_len, data, len); + r->cork_len += len; +} + static void scr_http_res_write_raw(ScrHttpRes *r, const char *data, size_t len) { if (r->finished) return; + if (r->corked > 0) { + scr_http_res_cork_buffer(r, data, len); + return; + } if (!r->head_sent) scr_http_res_send_head(r, -1); /* streaming: chunked */ if (r->h2_stream != NULL) { scr_http_h2_ops->write(r->h2_stream, data, len); @@ -790,11 +884,65 @@ void scr_http_res_write_bytes(ScrHttpRes *r, ScrBytes *data /*borrowed*/) { scr_http_res_write_raw(r, (const char *)data->data, data->len); } +/* res.flushHeaders(): the head goes out NOW (streaming framing — chunked + * unless the user fixed the length; the h2 lane's HEADERS frame). */ +void scr_http_res_flush_headers(ScrHttpRes *r) { + if (!r->head_sent && !r->finished) scr_http_res_send_head(r, -1); +} + +/* cork()/uncork(): the count IS res.writableCorked; the last uncork + * flushes the coalesced bytes as one write. */ +void scr_http_res_cork(ScrHttpRes *r) { r->corked++; } + +void scr_http_res_uncork(ScrHttpRes *r) { + if (r->corked == 0) return; + if (--r->corked == 0) scr_http_res_cork_flush(r); +} + +double scr_http_res_writable_corked(ScrHttpRes *r) { return (double)r->corked; } + +bool scr_http_res_destroyed_flag(ScrHttpRes *r) { return r->destroyed || r->close_emitted; } + +/* res.req — the paired request; a res.req = null write cleared it. */ +void scr_http_res_set_req(ScrHttpRes *r, ScrHttpReq *req /*borrowed*/) { + if (r->req_ref) scr_http_req_release(r->req_ref); + r->req_ref = req ? scr_http_req_retain(req) : NULL; +} + +/* res.setTimeout(ms[, cb]): the socket's idle timer, Node's delegation. */ +void scr_http_res_set_timeout(ScrHttpRes *r, double ms, ScrClosure *cb /*moves, nullable*/) { + if (r->finished || r->close_emitted || !r->sock) { + if (cb) scr_closure_release(cb); + return; + } + scr_net_sock_set_timeout(r->sock, ms); + if (cb) scr_net_sock_on_timeout(r->sock, cb, true); +} + static void scr_http_conn_response_finished(struct ScrHttpConn *conn, bool keep_alive); static void scr_http_queue_res_finish(ScrHttpRes *res); +/* Flush corked bytes as one write (corked already zero, or forced by + * end()). */ +static void scr_http_res_cork_flush(ScrHttpRes *r) { + if (r->cork_len == 0) return; + char *held = r->cork_buf; + size_t held_len = r->cork_len; + r->cork_buf = NULL; + r->cork_len = r->cork_cap = 0; + scr_http_res_write_raw(r, held, held_len); + free(held); +} + static void scr_http_res_end_raw(ScrHttpRes *r, const char *data, size_t len) { if (r->finished) return; + if (r->corked > 0 || r->cork_len > 0) { + /* end() flushes every cork level (Node) — the body streamed, so the + * framing below takes the already-committed streaming path */ + r->corked = 0; + scr_http_res_cork_flush(r); + if (r->finished) return; /* a teardown inside the flush */ + } if (r->h2_stream != NULL) { if (!r->head_sent) scr_http_res_send_head(r, (long long)len); scr_http_h2_ops->end(r->h2_stream, data, len); @@ -923,6 +1071,7 @@ void scr_http_res_write_head_pairs(ScrHttpRes *r, double status, ScrArr *pairs / * point — portless aborts a proxied response with it); 'close' follows * through the teardown path. Node-matched: no 'error' from a destroy. */ void scr_http_res_destroy(ScrHttpRes *r) { + r->destroyed = true; if (r->h2_stream != NULL) { /* h2 compat: destroy the STREAM (RST), never the shared session */ scr_http_h2_ops->destroy(r->h2_stream); @@ -956,7 +1105,9 @@ enum { SCR_HTTP_EMIT_REQ_CLOSE = 2, /* ScrHttpReq */ SCR_HTTP_EMIT_REQ_ABORTED = 3, /* ScrHttpReq: 'error' ("aborted") */ SCR_HTTP_EMIT_CLIENT_CLOSE = 4, /* ScrHttpClientReq */ - SCR_HTTP_EMIT_RES_FINISH = 5 /* ScrHttpRes: the end(cb) callbacks */ + SCR_HTTP_EMIT_RES_FINISH = 5, /* ScrHttpRes: the end(cb) callbacks */ + SCR_HTTP_EMIT_REQ_DRAIN = SCR_HTTP_EMIT_REQ_DRAIN_K, /* ScrHttpReq: resume()'s drain */ + SCR_HTTP_EMIT_RES_WCB = 7 /* ScrHttpRes: write(chunk, cb) callbacks */ }; typedef struct { @@ -1007,8 +1158,10 @@ static void scr_http_emit_push(int kind, void *h /*moves +1*/) { static void scr_http_emit_release(ScrHttpEmit *e) { switch (e->kind) { case SCR_HTTP_EMIT_RES_CLOSE: + case SCR_HTTP_EMIT_RES_WCB: case SCR_HTTP_EMIT_RES_FINISH: scr_http_res_release((ScrHttpRes *)e->h); break; case SCR_HTTP_EMIT_REQ_CLOSE: + case SCR_HTTP_EMIT_REQ_DRAIN: case SCR_HTTP_EMIT_REQ_ABORTED: scr_http_req_release((ScrHttpReq *)e->h); break; case SCR_HTTP_EMIT_CLIENT_CLOSE: scr_http_client_release_internal((struct ScrHttpClientReq *)e->h); @@ -1051,6 +1204,17 @@ void scr_http_res_on_finish(ScrHttpRes *r, ScrClosure *cb /*moves*/) { if (r->finished) scr_http_queue_res_finish(r); } +/* res.write(chunk, cb): the callback fires from the queue (this surface + * flushes synchronously into the socket buffer — the deferral keeps the + * cb off the writing stack, Node's contract). */ +void scr_http_res_on_write_flush(ScrHttpRes *r, ScrClosure *cb /*moves*/) { + scr_net_ls_add(&r->wcb_ls, cb, NULL, true); + if (!r->wcb_queued) { + r->wcb_queued = true; + scr_http_emit_push(SCR_HTTP_EMIT_RES_WCB, scr_http_res_retain(r)); + } +} + static bool scr_http_proto_pending(void) { return scr_http_emits_head < scr_http_emits_len; } static void scr_http_proto_sweep(void) { @@ -1061,15 +1225,23 @@ static void scr_http_proto_sweep(void) { ScrHttpRes *res = (ScrHttpRes *)e.h; if (!res->close_emitted) { res->close_emitted = true; + res->destroyed = true; /* Node: destroyed reads true inside 'close' */ scr_net_fire0_this(&res->close_ls, res, SCR_DYNH_HTTP_RES); scr_net_ls_drop(&res->close_ls); } break; } + case SCR_HTTP_EMIT_RES_WCB: { + ScrHttpRes *res = (ScrHttpRes *)e.h; + res->wcb_queued = false; /* later write(chunk, cb)s re-queue */ + scr_net_fire0_this(&res->wcb_ls, res, SCR_DYNH_HTTP_RES); + break; + } case SCR_HTTP_EMIT_REQ_CLOSE: { ScrHttpReq *req = (ScrHttpReq *)e.h; if (!req->close_emitted) { req->close_emitted = true; + req->destroyed = true; /* Node: destroyed reads true inside 'close' */ scr_net_fire0_this(&req->close_ls, req, SCR_DYNH_HTTP_REQ); scr_net_ls_drop(&req->err_ls); scr_net_ls_drop(&req->close_ls); @@ -1077,6 +1249,26 @@ static void scr_http_proto_sweep(void) { } break; } + case SCR_HTTP_EMIT_REQ_DRAIN: { + /* resume() after pause(): the held bytes deliver as one chunk (the + * reassembled body is the contract), then a deferred 'end'. A + * re-pause mid-drain re-buffers — the swap keeps the walk sound. */ + ScrHttpReq *req = (ScrHttpReq *)e.h; + req->drain_queued = false; + if (!req->paused) { + char *held = req->pend; + size_t held_len = req->pend_len; + req->pend = NULL; + req->pend_len = req->pend_cap = 0; + if (held_len > 0) scr_http_req_deliver(req, held, held_len); + free(held); + if (!scr_exc_pending() && !req->paused && req->end_pending) { + req->end_pending = false; + scr_http_req_finish(req, true); + } + } + break; + } case SCR_HTTP_EMIT_REQ_ABORTED: { ScrHttpReq *req = (ScrHttpReq *)e.h; if (!req->close_emitted) { @@ -1103,6 +1295,7 @@ static void scr_http_proto_sweep(void) { } static void scr_http_clients_cleanup(void); +static void scr_http_agents_cleanup(void); static void scr_http_cleanup_atexit(void) { scr_http_exiting = true; @@ -1113,6 +1306,7 @@ static void scr_http_cleanup_atexit(void) { free(scr_http_emits); scr_http_emits = NULL; scr_http_emits_head = scr_http_emits_len = scr_http_emits_cap = 0; + scr_http_agents_cleanup(); /* break agent↔client edges first */ scr_http_clients_cleanup(); } @@ -1274,6 +1468,20 @@ static void scr_http_conn_bad_request(ScrHttpConn *conn) { * feed, and the h2 compat DATA-frame feed). */ static void scr_http_req_deliver(ScrHttpReq *r, const char *data, size_t n) { if (!r || n == 0 || r->ended) return; + if (r->paused) { + /* req.pause(): the parser keeps consuming (memory is the buffer — + * a documented bound), delivery waits for resume()'s drain */ + if (r->pend_len + n > r->pend_cap) { + size_t cap = r->pend_cap ? r->pend_cap : 4096; + while (cap < r->pend_len + n) cap *= 2; + r->pend = realloc(r->pend, cap); + if (!r->pend) scr_http_oom(); + r->pend_cap = cap; + } + memcpy(r->pend + r->pend_len, data, n); + r->pend_len += n; + return; + } bool piped = r->pipe_res != NULL || r->pipe_client != NULL || r->pipe_sock != NULL; if (r->data_ls.n == 0 && !piped) return; ScrBytes *chunk = scr_bytes_new(SCR_BYTES_U8, (double)n); @@ -1495,6 +1703,7 @@ static bool scr_http_conn_parse_head(ScrHttpConn *conn, size_t head_len) { res->sock = scr_net_sock_retain(conn->sock); res->conn = conn; res->keep_alive = req_keep_alive; + res->req_ref = scr_http_req_retain(req); /* res.req — req never points back */ conn->req = req; /* conn's +1 */ conn->res = res; conn->close_after = !req_keep_alive; @@ -2037,6 +2246,9 @@ struct ScrHttpClientReq { bool close_emitted; /* settled: listeners dropped, off the registry */ ScrHttpReq *res; /* +1 once the head parses */ ScrNetLs resp_ls, err_ls, timeout_ls, close_ls, upgrade_ls; + /* the owning Agent (+1; the agent's entry holds this client +1 too — + * the cycle breaks at settle, or at the atexit agent sweep) */ + struct ScrHttpAgent *agent; bool in_registry; struct ScrHttpClientReq *next; }; @@ -2102,12 +2314,19 @@ static void scr_http_client_unregister(ScrHttpClientReq *c) { } } +/* Agent bookkeeping (implementation below, with the Agent unit): the + * settling client leaves its agent's lists and frees a slot. */ +static void scr_http_agent_client_done(struct ScrHttpAgent *ag, struct ScrHttpClientReq *c); +static struct ScrHttpAgent *scr_http_agent_release_p(struct ScrHttpAgent *ag); +static void scr_http_client_agent_detach(struct ScrHttpClientReq *c); + /* The queue's CLIENT_CLOSE emit: 'close' fires, the handle settles * (listeners drop — the cycle story) and leaves the registry. */ static void scr_http_client_settle(struct ScrHttpClientReq *c) { if (c->close_emitted) return; c->close_emitted = true; c->destroyed = true; + scr_http_client_agent_detach(c); /* usually already detached at socket close */ scr_net_fire0(&c->close_ls); scr_net_ls_drop(&c->resp_ls); scr_net_ls_drop(&c->err_ls); @@ -2558,9 +2777,22 @@ static void scr_http_client_eof(ScrHttpConn *conn) { scr_http_client_premature(conn); } +/* Detach a client from its agent (the socket died, or the handle + * settled): the entry leaves the lists BEFORE the socket's own 'close' + * listeners run — Node's agent removes its socket first too, so + * agent.sockets no longer names it inside a user 'close' handler. */ +static void scr_http_client_agent_detach(ScrHttpClientReq *c) { + if (!c->agent) return; + struct ScrHttpAgent *ag = c->agent; + c->agent = NULL; + scr_http_agent_client_done(ag, c); /* frees the slot; may dial the next */ + scr_http_agent_release_p(ag); +} + static void scr_http_client_closed(ScrHttpConn *conn) { ScrHttpClientReq *c = conn->client; if (!c) return; + scr_http_client_agent_detach(c); scr_http_client_premature(conn); /* the connection is gone: break the ctx→client edge (the client stays * alive through user refs / the queue until its 'close' settles) */ @@ -2744,6 +2976,300 @@ ScrHttpClientReq *scr_http_request(ScrStr *host /*borrowed*/, double port, fn, 80, NULL, NULL); } +/* ── the http Agent ────────────────────────────────────────────────────── + * + * Options, getName, destroy, and REAL maxSockets accounting over the + * one-dial-per-request connection model: an over-limit request's socket + * defers its dial (scr_net_connect_deferred) and queues; a dying + * active connection frees the slot and starts the next dial. What the + * runtime cannot express stays a NAMED fence: keep-alive socket POOLING + * (keepAlive: true) fences at construction — agent.freeSockets is + * honestly always empty, never a faked pool. + * + * Ownership: the agent registers (+1, atexit-swept); entries hold their + * client +1 and the client holds the agent +1 — the cycle breaks when + * the client's socket dies (scr_http_client_agent_detach) or at the + * atexit sweep. */ + +typedef struct ScrHttpAgentEnt { + ScrStr *name; /* getName's shape: "host:port:" */ + ScrHttpClientReq *client; /* +1 while listed */ + bool queued; /* waiting for a slot: the dial is deferred */ + struct ScrHttpAgentEnt *next; +} ScrHttpAgentEnt; + +typedef struct ScrHttpAgent { + size_t rc; + bool secure; /* https.Agent: protocol/defaultPort answers */ + bool keep_alive; /* always false here (true fences at construction) */ + double ka_msecs; + double max_sockets; /* INFINITY = Node's default */ + double max_free; + double timeout_ms; /* < 0 = unset */ + double default_port; /* settable (agent.defaultPort = p) */ + bool destroyed; + ScrHttpAgentEnt *ents; /* append order — Node's FIFO queue */ + bool in_registry; + struct ScrHttpAgent *next; +} ScrHttpAgent; + +static ScrHttpAgent *scr_http_agents = NULL; /* registry: +1 each, atexit-swept */ + +static ScrHttpAgent *scr_http_agent_retain(ScrHttpAgent *a) { + a->rc++; + return a; +} + +static void scr_http_agent_release(ScrHttpAgent *a) { + if (!a || --a->rc > 0) return; + ScrHttpAgentEnt *e = a->ents; + while (e) { + ScrHttpAgentEnt *next = e->next; + scr_str_release(e->name); + scr_http_client_release(e->client); + free(e); + e = next; + } + free(a); +} + +static struct ScrHttpAgent *scr_http_agent_release_p(struct ScrHttpAgent *ag) { + scr_http_agent_release(ag); + return NULL; +} + +static void *scr_http_agent_retain_v(void *p) { return scr_http_agent_retain((ScrHttpAgent *)p); } +static void scr_http_agent_release_v(void *p) { scr_http_agent_release((ScrHttpAgent *)p); } + +/* Actives (dialed, not settled) under a name. */ +static size_t scr_http_agent_active(const ScrHttpAgent *a, const ScrStr *name) { + size_t n = 0; + for (const ScrHttpAgentEnt *e = a->ents; e; e = e->next) { + if (!e->queued && e->name->len == name->len && + memcmp(e->name->data, name->data, name->len) == 0) n++; + } + return n; +} + +/* The settling/dying client leaves the lists; a freed slot starts the + * first queued dial for the same name (Node's FIFO). */ +static void scr_http_agent_client_done(struct ScrHttpAgent *ag, struct ScrHttpClientReq *c) { + ScrHttpAgentEnt **link = &ag->ents; + ScrStr *name = NULL; + while (*link && (*link)->client != c) link = &(*link)->next; + if (*link) { + ScrHttpAgentEnt *e = *link; + *link = e->next; + name = e->name; /* ownership moves out for the pump below */ + scr_http_client_release(e->client); + free(e); + } + if (name != NULL && !ag->destroyed) { + for (ScrHttpAgentEnt *e = ag->ents; e; e = e->next) { + if (e->queued && e->name->len == name->len && + memcmp(e->name->data, name->data, name->len) == 0) { + if (scr_http_agent_active(ag, name) < ag->max_sockets) { + e->queued = false; + if (e->client->sock) scr_net_sock_dial_start(e->client->sock); + } + break; + } + } + } + scr_str_release(name); +} + +/* getName's string builder — Node's exact shape: + * host:port:localAddress[:family][:socketPath]. */ +static ScrStr *scr_http_agent_name(const char *host, size_t host_len, const char *port, + size_t port_len, const char *laddr, size_t laddr_len, + int family, const char *spath, size_t spath_len) { + ScrJsonBuf b; + scr_jb_init(&b); + if (host_len == 0) { scr_jb_puts(&b, "localhost"); } + else for (size_t i = 0; i < host_len; i++) scr_jb_putc(&b, host[i]); + scr_jb_putc(&b, ':'); + for (size_t i = 0; i < port_len; i++) scr_jb_putc(&b, port[i]); + scr_jb_putc(&b, ':'); + for (size_t i = 0; i < laddr_len; i++) scr_jb_putc(&b, laddr[i]); + if (family == 4 || family == 6) { + scr_jb_putc(&b, ':'); + scr_jb_putc(&b, family == 4 ? '4' : '6'); + } + if (spath_len > 0) { + scr_jb_putc(&b, ':'); + for (size_t i = 0; i < spath_len; i++) scr_jb_putc(&b, spath[i]); + } + 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 */) { + if (keep_alive) { + static const char msg[] = + "an http Agent with keepAlive: true (socket pooling and reuse — compiled clients " + "dial one connection per request and close it with the response) is not supported " + "yet — construct the Agent without keepAlive, or drop the agent option"; + scr_throw_error_msg(SCR_ERR_ERROR, msg, sizeof msg - 1); + return NULL; + } + scr_http_install(); + ScrHttpAgent *a = calloc(1, sizeof *a); + if (!a) scr_http_oom(); + a->rc = 1; + a->secure = secure; + a->keep_alive = false; + a->ka_msecs = ka_msecs >= 0 ? ka_msecs : 1000; + a->max_sockets = max_sockets >= 0 ? max_sockets : (double)INFINITY; + a->max_free = max_free >= 0 ? max_free : 256; + a->timeout_ms = timeout_ms; + /* an agent-options `port` merges under portless request options (Node + * merges agent options into each connection) — the settable + * defaultPort carries it */ + a->default_port = port >= 0 ? port : secure ? 443 : 80; + /* registry (+1): the atexit sweep breaks agent↔client edges a program + * legitimately leaves in flight */ + a->in_registry = true; + a->next = scr_http_agents; + scr_http_agents = scr_http_agent_retain(a); + ScrDyn *d = scr_dyn_new_handle(a, SCR_DYNH_HTTP_AGENT); + scr_http_agent_release(a); /* the handle's retain holds it */ + return d; +} + +/* agent.destroy(): tears down every listed connection (actives destroy + * their sockets, queued dials never start) — Node destroys in-use + * sockets too; there is no free pool here. */ +static void scr_http_agent_destroy(ScrHttpAgent *a) { + a->destroyed = true; + /* destroying sockets detaches entries re-entrantly — walk a snapshot */ + for (;;) { + ScrHttpClientReq *victim = NULL; + for (ScrHttpAgentEnt *e = a->ents; e; e = e->next) { + if (e->client->sock && !e->client->close_emitted) { + victim = e->client; + break; + } + } + if (!victim) break; + scr_http_client_agent_detach(victim); /* leaves the list first */ + scr_net_sock_destroy(victim->sock); + } +} + +ScrHttpClientReq *scr_http_request_agent_ex(ScrStr *host /*borrowed*/, double port, + ScrStr *path /*borrowed*/, ScrStr *method /*borrowed*/, + double timeout_ms, ScrArr *header_pairs /*borrowed*/, + bool auto_end, const ScrDyn *agent /*borrowed*/, + ScrClosure *cb /*moves, nullable*/, + ScrHttpRespFn fn, int default_port, + void (*wrap)(ScrNetSocket *, void *), void *wrap_ctx) { + /* the null/undefined arms: the default path (port < 0 = no port option) */ + if (agent == NULL || agent->kind == SCR_DYN_UNDEF || agent->kind == SCR_DYN_NULL) { + double p = port >= 0 ? port : (double)default_port; + return scr_http_request_impl(host, p, path, method, timeout_ms, header_pairs, auto_end, + cb, fn, default_port, wrap, wrap_ctx, NULL); + } + /* agent: false — Node's one-shot Agent: exactly this client's model, + * plus its Connection: close request header (unless the caller set one) */ + if (agent->kind == SCR_DYN_BOOL && !agent->v.b) { + double p = port >= 0 ? port : (double)default_port; + bool has_conn = false; + size_t npairs = (size_t)scr_arr_len(header_pairs); + for (size_t i = 0; i + 1 < npairs && !has_conn; i += 2) { + ScrStr *n = (ScrStr *)scr_arr_get_ref(header_pairs, (double)i); + if (n->len == 10) { + bool eq = true; + for (size_t j = 0; j < 10 && eq; j++) { + if (tolower((unsigned char)n->data[j]) != "connection"[j]) eq = false; + } + has_conn = eq; + } + scr_str_release(n); + } + ScrArr *pairs = header_pairs; + ScrArr *copy = NULL; + if (!has_conn) { + copy = scr_arr_new(SCR_ELEM_STR, npairs + 2); + for (size_t i = 0; i < npairs; i++) scr_arr_push_ref(copy, scr_arr_get_ref(header_pairs, (double)i)); + scr_arr_push_ref(copy, scr_str_new("Connection", 10)); + scr_arr_push_ref(copy, scr_str_new("close", 5)); + pairs = copy; + } + ScrHttpClientReq *c = scr_http_request_impl(host, p, path, method, timeout_ms, pairs, + auto_end, cb, fn, default_port, wrap, wrap_ctx, NULL); + if (copy) scr_arr_release(copy); + return c; + } + if (agent->kind != SCR_DYN_HANDLE || agent->v.handle.tag != SCR_DYNH_HTTP_AGENT) { + if (cb) scr_closure_release(cb); + 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, NULL, 0); + bool queue = scr_http_agent_active(ag, name) >= ag->max_sockets; + ScrNetSocket *presock = queue ? 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 */ + scr_str_release(name); + return NULL; + } + if (ag->timeout_ms >= 0 && timeout_ms <= 0) scr_net_sock_set_timeout(c->sock, ag->timeout_ms); + 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(ScrStr *host /*borrowed*/, double port, + ScrStr *path /*borrowed*/, ScrStr *method /*borrowed*/, + double timeout_ms, ScrArr *header_pairs /*borrowed*/, + bool auto_end, const ScrDyn *agent /*borrowed*/, + ScrClosure *cb /*moves, nullable*/, ScrHttpRespFn fn) { + return scr_http_request_agent_ex(host, port, path, method, timeout_ms, header_pairs, auto_end, + agent, cb, fn, 80, NULL, NULL); +} + +/* The atexit sweep: break agent↔client edges a program leaves in flight + * (queued dials that never ran) so the RC audit sees a clean heap. */ +static void scr_http_agents_cleanup(void) { + while (scr_http_agents) { + ScrHttpAgent *a = scr_http_agents; + scr_http_agents = a->next; + a->in_registry = false; + ScrHttpAgentEnt *e = a->ents; + a->ents = NULL; + while (e) { + ScrHttpAgentEnt *next = e->next; + if (e->client->agent) { + scr_http_agent_release(e->client->agent); + e->client->agent = NULL; + } + scr_str_release(e->name); + scr_http_client_release(e->client); + free(e); + e = next; + } + scr_http_agent_release(a); + } +} + /* Exit-time registry cleanup (the net-unit precedent): clients a program * legitimately leaves in flight at exit release their listeners and * registry references so the RC audit sees a clean heap. */ @@ -2854,12 +3380,43 @@ static bool scr_http_dynh_chunk_ok(const ScrDyn *chunk) { return false; } -/* An optional trailing encoding argument: utf8 spellings pass (chunks - * already carry utf8 bytes), anything else fences loudly. */ -static bool scr_http_dynh_enc_ok(const ScrDyn *enc, const char *cls, const char *member) { - if (scr_http_dynh_name_is(enc, "utf8") || scr_http_dynh_name_is(enc, "utf-8")) return true; - scr_http_dynh_unsupported(cls, member, "only the 'utf8' encoding is modeled"); - return false; +/* An optional trailing encoding argument on write/end: utf8 spellings + * pass through (string chunks already carry utf8 bytes); every other + * real Node encoding decodes the STRING chunk through Buffer.from's + * decoder into *out (+1; buffer chunks ignore the encoding, Node); + * unknown names throw Node's ERR_UNKNOWN_ENCODING. False = exception + * pending. */ +static bool scr_http_dynh_encode(const ScrDyn *chunk, const ScrDyn *enc, ScrBytes **out) { + *out = NULL; + if (enc->kind != SCR_DYN_STR || !scr_bytes_is_encoding(enc->v.str)) { + ScrJsonBuf b; + scr_jb_init(&b); + scr_jb_puts(&b, "Unknown encoding: "); + if (enc->kind == SCR_DYN_STR) { + for (size_t i = 0; i < enc->v.str->len && i < 64; i++) scr_jb_putc(&b, enc->v.str->data[i]); + } + ScrStr *msg = scr_jb_finish(&b); + scr_throw_error_msg_code(SCR_ERR_TYPE, msg->data, msg->len, "ERR_UNKNOWN_ENCODING"); + scr_str_release(msg); + return false; + } + if (chunk->kind != SCR_DYN_STR) return true; /* buffers carry their bytes */ + /* lowercase + alias normalization down to scr_bytes_from_str's arms */ + char low[10]; + size_t n = enc->v.str->len < 9 ? enc->v.str->len : 9; + for (size_t i = 0; i < n; i++) { + char c = enc->v.str->data[i]; + low[i] = c >= 'A' && c <= 'Z' ? (char)(c + 32) : c; + } + low[n] = 0; + const char *canon = low; + if (strcmp(low, "utf8") == 0 || strcmp(low, "utf-8") == 0) return true; /* passthrough */ + if (strcmp(low, "binary") == 0) canon = "latin1"; + else if (strcmp(low, "ucs2") == 0 || strcmp(low, "ucs-2") == 0 || strcmp(low, "utf-16le") == 0) canon = "utf16le"; + ScrStr *cs = scr_str_new(canon, strlen(canon)); + *out = scr_bytes_from_str(chunk->v.str, cs); + scr_str_release(cs); + return true; } /* ── IncomingMessage (SCR_DYNH_HTTP_REQ) ─────────────────────────────── */ @@ -2883,6 +3440,9 @@ static ScrDyn *scr_http_dynh_req_invoke(void *h, ScrDyn *self, const char *metho scr_http_req_on_close(r, scr_dyn_listener_closure0(cb), once); } else if (scr_http_dynh_name_is(name, "aborted")) { scr_http_req_on_aborted(r, scr_dyn_listener_closure0(cb), once); + } else if (scr_http_dynh_name_is(name, "timeout") && r->sock != NULL) { + /* the socket's idle timer — req.setTimeout's event, Node's delegation */ + scr_net_sock_on_timeout(r->sock, scr_dyn_listener_closure0(cb), once); } else { scr_http_dynh_event_unsupported("IncomingMessage", name); return NULL; @@ -2893,6 +3453,25 @@ static ScrDyn *scr_http_dynh_req_invoke(void *h, ScrDyn *self, const char *metho scr_http_req_resume(r); return scr_dyn_retain(self); } + if (strcmp(method, "pause") == 0) { + scr_http_req_pause(r); + return scr_dyn_retain(self); + } + if (strcmp(method, "setTimeout") == 0) { + const ScrDyn *ms = argc > 0 ? args[0] : scr_dyn_undefined(); + if (ms->kind != SCR_DYN_NUM) { + scr_dyn_arg_type_fail("msecs", "of type number", ms); + return NULL; + } + ScrClosure *cb = NULL; + if (argc > 1 && args[1]->kind == SCR_DYN_FUNC) cb = scr_dyn_listener_closure0(args[1]); + else if (argc > 1 && args[1]->kind != SCR_DYN_UNDEF) { + scr_dyn_check_listener(args[1], "callback"); + return NULL; + } + scr_http_req_set_timeout(r, ms->v.num, cb); + return scr_dyn_retain(self); + } if (strcmp(method, "setEncoding") == 0) { const ScrDyn *enc = argc > 0 ? args[0] : scr_dyn_undefined(); if (enc->kind != SCR_DYN_STR) { @@ -2932,7 +3511,7 @@ static ScrDyn *scr_http_dynh_req_invoke(void *h, ScrDyn *self, const char *metho } { /* Real prototype members without a modeled dispatch: loud. */ - static const char *const known[] = { "pause", "unpipe", "setTimeout", + static const char *const known[] = { "unpipe", "read", "push", "off", "removeListener", "removeAllListeners", "emit", "prependListener", "prependOnceListener", "listenerCount", "listeners", "isPaused", "wrap", "cork", "uncork", NULL }; @@ -3028,12 +3607,15 @@ static ScrDyn *scr_http_dynh_req_get(void *h, const char *key, size_t key_len) { } if (strcmp(key, "aborted") == 0) return scr_dyn_new_bool(r->aborted); if (strcmp(key, "complete") == 0) return scr_dyn_new_bool(r->ended); + if (strcmp(key, "destroyed") == 0) return scr_dyn_new_bool(scr_http_req_destroyed_flag(r)); + if (strcmp(key, "readable") == 0) return scr_dyn_new_bool(scr_http_req_readable(r)); + if (strcmp(key, "readableEnded") == 0) return scr_dyn_new_bool(r->ended); + if (strcmp(key, "closed") == 0) return scr_dyn_new_bool(r->close_emitted); { /* Real instance properties without a modeled read: loud, never a * silent undefined where Node has a value. */ static const char *const known[] = { - "trailers", "rawTrailers", "readable", "readableEnded", - "closed", "destroyed", NULL }; + "trailers", "rawTrailers", NULL }; if (scr_http_dynh_in(key, known)) { scr_http_dynh_unsupported("IncomingMessage", key, NULL); return NULL; @@ -3088,6 +3670,55 @@ static ScrArr *scr_http_dynh_pairs(const ScrDyn *headers) { return pairs; } +/* writeHead's RAW-array form over a DOM ARR: [k0, v0, k1, v1, ...] + * flattens into the same pairs feed; an odd length throws Node's + * ERR_INVALID_ARG_VALUE; value slots may be arrays (consecutive + * same-name lines) or numbers (String(n)). NULL = exception pending. */ +static ScrArr *scr_http_dynh_flat_pairs(const ScrDyn *list) { + size_t n = list->v.arr.len; + if (n % 2 != 0) { + static const char msg[] = "The argument 'headers' is invalid."; + scr_throw_error_msg_code(SCR_ERR_TYPE, msg, sizeof msg - 1, "ERR_INVALID_ARG_VALUE"); + return NULL; + } + ScrArr *pairs = scr_arr_new(SCR_ELEM_STR, n); + for (size_t i = 0; i + 1 < n; i += 2) { + const ScrDyn *k = list->v.arr.items[i]; + const ScrDyn *v = list->v.arr.items[i + 1]; + if (k->kind != SCR_DYN_STR) { + scr_arr_release(pairs); + scr_dyn_arg_type_fail("name", "of type string", k); + return NULL; + } + size_t reps = v->kind == SCR_DYN_ARR ? v->v.arr.len : 1; + for (size_t rep = 0; rep < reps; rep++) { + const ScrDyn *one = v->kind == SCR_DYN_ARR ? v->v.arr.items[rep] : v; + ScrStr *vs; + if (one->kind == SCR_DYN_STR) { + vs = scr_str_retain(one->v.str); + } else if (one->kind == SCR_DYN_NUM) { + vs = scr_f64_to_scrstr(one->v.num); + } else { + ScrJsonBuf b; + scr_jb_init(&b); + scr_jb_puts(&b, "Invalid value \""); + scr_jb_puts(&b, one->kind == SCR_DYN_UNDEF ? "undefined" : "object"); + scr_jb_puts(&b, "\" for header \""); + for (size_t c = 0; c < k->v.str->len; c++) scr_jb_putc(&b, k->v.str->data[c]); + scr_jb_putc(&b, '"'); + ScrStr *msg = scr_jb_finish(&b); + scr_throw_error_msg_code(SCR_ERR_TYPE, msg->data, msg->len, "ERR_HTTP_INVALID_HEADER_VALUE"); + scr_str_release(msg); + scr_arr_release(pairs); + return NULL; + } + scr_arr_push_ref(pairs, scr_str_new(k->v.str->data, k->v.str->len)); + scr_arr_push_ref(pairs, vs); + } + } + return pairs; +} + static ScrDyn *scr_http_dynh_res_invoke(void *h, ScrDyn *self, const char *method, ScrDyn *const *args, size_t argc, const char *what) { ScrHttpRes *r = (ScrHttpRes *)h; @@ -3105,6 +3736,9 @@ static ScrDyn *scr_http_dynh_res_invoke(void *h, ScrDyn *self, const char *metho /* This surface never backpressures (write answers true), so Node's * contract says 'drain' never fires — an accepted, never-fired * registration is the consistent answer (SEMANTICS.md). */ + } else if (scr_http_dynh_name_is(name, "timeout") && r->sock != NULL) { + /* the socket's idle timer — res.setTimeout's event, Node's delegation */ + scr_net_sock_on_timeout(r->sock, scr_dyn_listener_closure0(cb), once); } else { scr_http_dynh_event_unsupported("ServerResponse", name); return NULL; @@ -3117,41 +3751,117 @@ static ScrDyn *scr_http_dynh_res_invoke(void *h, ScrDyn *self, const char *metho size_t i = 0; const ScrDyn *chunk = NULL; const ScrDyn *cb = NULL; + ScrBytes *decoded = NULL; if (i < argc && args[i]->kind == SCR_DYN_FUNC) { cb = args[i++]; } else if (i < argc && args[i]->kind != SCR_DYN_UNDEF && args[i]->kind != SCR_DYN_NULL) { if (!scr_http_dynh_chunk_ok(args[i])) return NULL; chunk = args[i++]; if (i < argc && args[i]->kind == SCR_DYN_STR) { /* encoding */ - if (!scr_http_dynh_enc_ok(args[i], "ServerResponse", "end")) return NULL; + if (!scr_http_dynh_encode(chunk, args[i], &decoded)) return NULL; i++; } if (i < argc && args[i]->kind == SCR_DYN_FUNC) cb = args[i++]; } if (i < argc && args[i]->kind != SCR_DYN_UNDEF) { + scr_bytes_release(decoded); scr_http_dynh_unsupported("ServerResponse", "end", "this argument shape is not modeled"); return NULL; } if (cb) scr_http_res_on_finish(r, scr_dyn_listener_closure0(cb)); - if (chunk == NULL) scr_http_res_end(r); - else if (chunk->kind == SCR_DYN_STR) scr_http_res_end_str(r, chunk->v.str); - else scr_http_res_end_bytes(r, chunk->v.bytes); + if (decoded != NULL) { + scr_http_res_end_bytes(r, decoded); + scr_bytes_release(decoded); + } else if (chunk == NULL) { + scr_http_res_end(r); + } else if (chunk->kind == SCR_DYN_STR) { + scr_http_res_end_str(r, chunk->v.str); + } else { + scr_http_res_end_bytes(r, chunk->v.bytes); + } return scr_dyn_retain(self); } if (strcmp(method, "write") == 0) { const ScrDyn *chunk = argc > 0 ? args[0] : scr_dyn_undefined(); if (!scr_http_dynh_chunk_ok(chunk)) return NULL; - if (argc > 1 && args[1]->kind == SCR_DYN_STR && !scr_http_dynh_enc_ok(args[1], "ServerResponse", "write")) return NULL; - if (argc > 1 && args[1]->kind == SCR_DYN_FUNC) { - scr_http_dynh_unsupported("ServerResponse", "write", "write(chunk, callback) is not modeled"); - return NULL; + ScrBytes *decoded = NULL; + size_t i = 1; + if (i < argc && args[i]->kind == SCR_DYN_STR) { /* encoding */ + if (!scr_http_dynh_encode(chunk, args[i], &decoded)) return NULL; + i++; + } + if (i < argc && args[i]->kind == SCR_DYN_FUNC) { + /* write(chunk[, enc], cb): fires from the queue once the chunk + * entered the socket buffer — this surface's flush moment */ + scr_http_res_on_write_flush(r, scr_dyn_listener_closure0(args[i])); + i++; + } + if (decoded != NULL) { + scr_http_res_write_bytes(r, decoded); + scr_bytes_release(decoded); + } else if (chunk->kind == SCR_DYN_STR) { + scr_http_res_write_str(r, chunk->v.str); + } else { + scr_http_res_write_bytes(r, chunk->v.bytes); } - if (chunk->kind == SCR_DYN_STR) scr_http_res_write_str(r, chunk->v.str); - else scr_http_res_write_bytes(r, chunk->v.bytes); /* Always-true: this surface buffers without a highWaterMark verdict * (SEMANTICS.md — backpressure is not modeled). */ return scr_dyn_new_bool(true); } + if (strcmp(method, "flushHeaders") == 0) { + scr_http_res_flush_headers(r); + return scr_dyn_retain(scr_dyn_undefined()); + } + if (strcmp(method, "cork") == 0) { + scr_http_res_cork(r); + return scr_dyn_retain(scr_dyn_undefined()); + } + if (strcmp(method, "uncork") == 0) { + scr_http_res_uncork(r); + return scr_dyn_retain(scr_dyn_undefined()); + } + if (strcmp(method, "setTimeout") == 0) { + const ScrDyn *ms = argc > 0 ? args[0] : scr_dyn_undefined(); + if (ms->kind != SCR_DYN_NUM) { + scr_dyn_arg_type_fail("msecs", "of type number", ms); + return NULL; + } + ScrClosure *tcb = NULL; + if (argc > 1 && args[1]->kind == SCR_DYN_FUNC) tcb = scr_dyn_listener_closure0(args[1]); + else if (argc > 1 && args[1]->kind != SCR_DYN_UNDEF) { + scr_dyn_check_listener(args[1], "callback"); + return NULL; + } + scr_http_res_set_timeout(r, ms->v.num, tcb); + return scr_dyn_retain(self); + } + if (strcmp(method, "getHeaders") == 0) { + /* the snapshot object — lowercased names; repeated names read as an + * array in set order (Node's outgoing shape) */ + ScrDyn *obj = scr_dyn_new_obj(); + for (size_t k = 0; k < r->nheaders; k++) { + const ScrStr *n = r->hnames[k]; + char lower[256]; + size_t nl = n->len < sizeof lower ? n->len : sizeof lower - 1; + for (size_t c = 0; c < nl; c++) { + char ch = n->data[c]; + lower[c] = ch >= 'A' && ch <= 'Z' ? (char)(ch + 32) : ch; + } + ScrDyn *prev = scr_dyn_obj_get(obj, lower, nl); + ScrDyn *val = scr_dyn_new_str(r->hvalues[k]); + if (prev == NULL) { + scr_dyn_obj_set(obj, lower, nl, val); /* moves */ + } else if (prev->kind == SCR_DYN_ARR) { + scr_dyn_arr_push(prev, val); /* moves */ + } else { + ScrDyn *arr = scr_dyn_new_arr(); + scr_dyn_arr_push(arr, scr_dyn_retain(prev)); + scr_dyn_arr_push(arr, val); + scr_dyn_obj_set(obj, lower, nl, arr); /* moves; releases prev */ + } + } + return obj; + } if (strcmp(method, "writeHead") == 0) { const ScrDyn *st = argc > 0 ? args[0] : scr_dyn_undefined(); if (st->kind != SCR_DYN_NUM) { @@ -3169,8 +3879,21 @@ static ScrDyn *scr_http_dynh_res_invoke(void *h, ScrDyn *self, const char *metho scr_http_res_write_head_pairs(r, st->v.num, pairs); scr_arr_release(pairs); i++; + } else if (i < argc && args[i]->kind == SCR_DYN_ARR) { + /* the RAW-array form: [k0, v0, k1, v1, ...] — an even length is + * Node's contract (ERR_INVALID_ARG_VALUE otherwise), values may be + * arrays (consecutive same-name lines, the setHeader expansion), + * and per NAME the list overrides what setHeader() stored while + * other names survive — the object form's writeHead-wins merge + * (oracle-pinned: a res with setHeader('a') plus writeHead(200, + * ['test', ...]) sends both). */ + ScrArr *pairs = scr_http_dynh_flat_pairs(args[i]); + if (!pairs) return NULL; + scr_http_res_write_head_pairs(r, st->v.num, pairs); + scr_arr_release(pairs); + i++; } else if (i < argc && args[i]->kind != SCR_DYN_UNDEF) { - scr_http_dynh_unsupported("ServerResponse", "writeHead", "only the header-object form is modeled"); + scr_http_dynh_unsupported("ServerResponse", "writeHead", "only the header-object and raw-array forms are modeled"); return NULL; } else { scr_http_res_write_head(r, st->v.num); @@ -3250,8 +3973,8 @@ static ScrDyn *scr_http_dynh_res_invoke(void *h, ScrDyn *self, const char *metho return d; } { - static const char *const known[] = { "writeContinue", "writeEarlyHints", "cork", "uncork", - "flushHeaders", "getHeaders", "getHeaderNames", "appendHeader", "setTimeout", + static const char *const known[] = { "writeContinue", "writeEarlyHints", + "getHeaderNames", "appendHeader", "addTrailers", "off", "removeListener", "removeAllListeners", "emit", "prependListener", "prependOnceListener", "listenerCount", "listeners", "setDefaultEncoding", NULL }; if (scr_http_dynh_in(method, known)) { @@ -3283,10 +4006,23 @@ static ScrDyn *scr_http_dynh_res_get(void *h, const char *key, size_t key_len) { if (!r->sock) return scr_dyn_new_null(); /* destroyed: Node nulls it */ return scr_dyn_new_handle(r->sock, SCR_DYNH_NET_SOCKET); } + if (strcmp(key, "req") == 0) { + if (r->req_cleared || r->req_ref == NULL) return scr_dyn_new_null(); + return scr_dyn_new_handle(r->req_ref, SCR_DYNH_HTTP_REQ); + } + if (strcmp(key, "writableCorked") == 0) return scr_dyn_new_num(scr_http_res_writable_corked(r)); + if (strcmp(key, "writableFinished") == 0) return scr_dyn_new_bool(r->finished); + if (strcmp(key, "writableHighWaterMark") == 0) { + /* Node's default socket highWaterMark — a constant here (backpressure + * is not modeled; SEMANTICS.md) */ + return scr_dyn_new_num(16384); + } + if (strcmp(key, "destroyed") == 0) return scr_dyn_new_bool(scr_http_res_destroyed_flag(r)); + if (strcmp(key, "closed") == 0) return scr_dyn_new_bool(r->close_emitted); { - static const char *const known[] = { "req", "chunkedEncoding", "sendDate", - "strictContentLength", "writableFinished", "writableLength", "writableHighWaterMark", - "writableCorked", "writableObjectMode", "closed", "destroyed", "errored", NULL }; + static const char *const known[] = { "chunkedEncoding", "sendDate", + "strictContentLength", "writableLength", + "writableObjectMode", "errored", NULL }; if (scr_http_dynh_in(key, known)) { scr_http_dynh_unsupported("ServerResponse", key, NULL); return NULL; @@ -3320,6 +4056,14 @@ static bool scr_http_dynh_res_set(void *h, const char *key, size_t key_len, cons r->no_date = !scr_dyn_truthy(value); return true; } + if (strcmp(key, "req") == 0 && + (value->kind == SCR_DYN_NULL || value->kind == SCR_DYN_UNDEF)) { + /* res.req is a plain property in Node — a null/undefined write + * clears the read; other writes keep the loud fence (the handle + * cannot carry arbitrary values). */ + r->req_cleared = true; + return true; + } return false; } @@ -3392,9 +4136,311 @@ static bool scr_http_dynh_server_on(ScrNetServer *s, const char *event, const Sc return false; } +/* ── ClientRequest (SCR_DYNH_HTTP_CLIENT) ────────────────────────────── */ + +/* The 'response' fire for a DYN listener (the reqres adapter's shape): + * res arrives +1, boxes, and the checked-dynamic call runs. */ +static void scr_http_dynh_fire_resp(ScrClosure *cb, ScrHttpReq *res /* +1 */) { + ScrDyn *fn = scr_dyn_listener_fn(cb); + ScrDyn *dres = scr_dyn_new_handle(res, SCR_DYNH_HTTP_REQ); + ScrDyn *args1[1] = { dres }; + ScrDyn *r = scr_dyn_call(fn, args1, 1, "listener"); + scr_dyn_release(r); + scr_dyn_release(dres); + scr_dyn_release(fn); + scr_http_req_release(res); +} + +static ScrDyn *scr_http_dynh_client_invoke(void *h, ScrDyn *self, const char *method, + ScrDyn *const *args, size_t argc, const char *what) { + ScrHttpClientReq *c = (ScrHttpClientReq *)h; + bool once = false; + if (scr_http_dynh_reg(method, &once)) { + const ScrDyn *name = argc > 0 ? args[0] : scr_dyn_undefined(); + const ScrDyn *cb = argc > 1 ? args[1] : scr_dyn_undefined(); + scr_dyn_check_listener(cb, "listener"); + if (scr_exc_pending()) return NULL; + if (scr_http_dynh_name_is(name, "response")) { + scr_http_client_on_response(c, scr_dyn_listener_closure_fn(cb, (void *)&scr_http_dynh_fire_resp), + (ScrHttpRespFn)&scr_http_dynh_fire_resp, once); + } else if (scr_http_dynh_name_is(name, "error")) { + scr_http_client_on_error(c, scr_dyn_listener_closure_err(cb), (ScrChildErrFn)&scr_dyn_listener_fire_err, once); + } else if (scr_http_dynh_name_is(name, "close")) { + scr_http_client_on_close(c, scr_dyn_listener_closure0(cb), once); + } else if (scr_http_dynh_name_is(name, "timeout")) { + scr_http_client_on_timeout(c, scr_dyn_listener_closure0(cb), once); + } else { + scr_http_dynh_event_unsupported("ClientRequest", name); + return NULL; + } + return scr_dyn_retain(self); + } + if (strcmp(method, "end") == 0) { + const ScrDyn *chunk = argc > 0 ? args[0] : scr_dyn_undefined(); + if (chunk->kind == SCR_DYN_STR) scr_http_client_end_str(c, chunk->v.str); + else if (chunk->kind == SCR_DYN_BYTES) scr_http_client_end_bytes(c, chunk->v.bytes); + else if (chunk->kind == SCR_DYN_UNDEF || chunk->kind == SCR_DYN_NULL) scr_http_client_end(c); + else { + scr_dyn_arg_type_fail("chunk", "of type string or an instance of Buffer or Uint8Array", chunk); + return NULL; + } + return scr_dyn_retain(self); + } + if (strcmp(method, "write") == 0) { + const ScrDyn *chunk = argc > 0 ? args[0] : scr_dyn_undefined(); + if (!scr_http_dynh_chunk_ok(chunk)) return NULL; + if (chunk->kind == SCR_DYN_STR) scr_http_client_write_str(c, chunk->v.str); + else scr_http_client_write_bytes(c, chunk->v.bytes); + return scr_dyn_new_bool(true); + } + if (strcmp(method, "destroy") == 0) { + scr_http_client_destroy(c); + return scr_dyn_retain(self); + } + if (strcmp(method, "setTimeout") == 0) { + const ScrDyn *ms = argc > 0 ? args[0] : scr_dyn_undefined(); + if (ms->kind != SCR_DYN_NUM) { + scr_dyn_arg_type_fail("msecs", "of type number", ms); + return NULL; + } + scr_http_client_set_timeout(c, ms->v.num); + if (argc > 1 && args[1]->kind == SCR_DYN_FUNC) { + scr_http_client_on_timeout(c, scr_dyn_listener_closure0(args[1]), true); + } + return scr_dyn_retain(self); + } + if (strcmp(method, "flushHeaders") == 0) { + if (!c->head_sent && !c->destroyed) scr_http_client_send_head(c, -1); + return scr_dyn_retain(scr_dyn_undefined()); + } + if (strcmp(method, "toString") == 0) { + ScrStr *s = scr_str_new("[object Object]", 15); + ScrDyn *d = scr_dyn_new_str(s); + scr_str_release(s); + return d; + } + { + static const char *const known[] = { "abort", "setNoDelay", "setSocketKeepAlive", + "getHeader", "setHeader", "removeHeader", "getHeaders", "getHeaderNames", "hasHeader", + "cork", "uncork", "pipe", "off", "removeListener", "removeAllListeners", "emit", + "prependListener", "prependOnceListener", "listenerCount", "listeners", NULL }; + if (scr_http_dynh_in(method, known)) { + scr_http_dynh_unsupported("ClientRequest", method, NULL); + return NULL; + } + } + scr_http_dynh_not_fn(what); + return NULL; +} + +static ScrDyn *scr_http_dynh_client_get(void *h, const char *key, size_t key_len) { + ScrHttpClientReq *c = (ScrHttpClientReq *)h; + (void)key_len; + if (strcmp(key, "socket") == 0 || strcmp(key, "connection") == 0) { + if (!c->sock) return scr_dyn_new_null(); + return scr_dyn_new_handle(c->sock, SCR_DYNH_NET_SOCKET); + } + if (strcmp(key, "destroyed") == 0) return scr_dyn_new_bool(scr_http_client_destroyed(c)); + if (strcmp(key, "writableEnded") == 0) return scr_dyn_new_bool(c->ended); + if (strcmp(key, "path") == 0) { + ScrDyn *d = scr_dyn_new_str(c->path); + return d; + } + if (strcmp(key, "method") == 0) return scr_dyn_new_str(c->method); + { + static const char *const known[] = { "aborted", "host", "protocol", "res", + "reusedSocket", "maxHeadersCount", "headersSent", "writableFinished", NULL }; + if (scr_http_dynh_in(key, known)) { + scr_http_dynh_unsupported("ClientRequest", key, NULL); + return NULL; + } + } + return NULL; +} + +static bool scr_http_dynh_client_set(void *h, const char *key, size_t key_len, const ScrDyn *value) { + (void)h; (void)key; (void)key_len; (void)value; + return false; +} + +static const ScrDynHandleOps scr_http_dynh_client_ops = { + "ClientRequest", + &scr_http_client_retain_v, + &scr_http_client_release_v, + &scr_http_dynh_client_invoke, + &scr_http_dynh_client_get, + &scr_http_dynh_client_set, + NULL, +}; + +/* ── Agent (SCR_DYNH_HTTP_AGENT) ─────────────────────────────────────── */ + +/* A string-ish option field out of a DOM OBJ: STR answers its bytes, + * NUM formats (the port), everything else reads as absent. */ +static ScrStr *scr_http_dynh_opt_str(const ScrDyn *opts, const char *key) { + if (opts == NULL || opts->kind != SCR_DYN_OBJ) return NULL; + const ScrDyn *v = scr_dyn_obj_get(opts, key, strlen(key)); + if (v == NULL) return NULL; + if (v->kind == SCR_DYN_STR) return scr_str_retain(v->v.str); + if (v->kind == SCR_DYN_NUM) return scr_f64_to_scrstr(v->v.num); + return NULL; +} + +static ScrDyn *scr_http_dynh_agent_invoke(void *h, ScrDyn *self, const char *method, + ScrDyn *const *args, size_t argc, const char *what) { + ScrHttpAgent *a = (ScrHttpAgent *)h; + bool once = false; + if (scr_http_dynh_reg(method, &once)) { + /* Agent events ('free'/'timeout') ride the pooling the runtime does + * not have — loud, never a silently-dead listener. */ + const ScrDyn *name = argc > 0 ? args[0] : scr_dyn_undefined(); + scr_http_dynh_event_unsupported("Agent", name); + return NULL; + } + if (strcmp(method, "getName") == 0) { + const ScrDyn *opts = argc > 0 ? args[0] : scr_dyn_undefined(); + ScrStr *host = scr_http_dynh_opt_str(opts, "host"); + ScrStr *port = scr_http_dynh_opt_str(opts, "port"); + ScrStr *laddr = scr_http_dynh_opt_str(opts, "localAddress"); + ScrStr *spath = scr_http_dynh_opt_str(opts, "socketPath"); + int family = 0; + if (opts != NULL && opts->kind == SCR_DYN_OBJ) { + const ScrDyn *f = scr_dyn_obj_get(opts, "family", 6); + if (f != NULL && f->kind == SCR_DYN_NUM && (f->v.num == 4 || f->v.num == 6)) { + family = (int)f->v.num; + } + } + ScrStr *name = scr_http_agent_name(host ? host->data : NULL, host ? host->len : 0, + port ? port->data : NULL, port ? port->len : 0, + laddr ? laddr->data : NULL, laddr ? laddr->len : 0, + family, spath ? spath->data : NULL, + spath ? spath->len : 0); + scr_str_release(host); + scr_str_release(port); + scr_str_release(laddr); + scr_str_release(spath); + ScrDyn *d = scr_dyn_new_str(name); + scr_str_release(name); + return d; + } + if (strcmp(method, "destroy") == 0) { + scr_http_agent_destroy(a); + return scr_dyn_retain(scr_dyn_undefined()); + } + if (strcmp(method, "toString") == 0) { + ScrStr *s = scr_str_new("[object Object]", 15); + ScrDyn *d = scr_dyn_new_str(s); + scr_str_release(s); + return d; + } + { + static const char *const known[] = { "createConnection", "createSocket", "keepSocketAlive", + "reuseSocket", "addRequest", "removeSocket", "off", "removeListener", + "removeAllListeners", "emit", "prependListener", "prependOnceListener", + "listenerCount", "listeners", NULL }; + if (scr_http_dynh_in(method, known)) { + scr_http_dynh_unsupported("Agent", method, NULL); + return NULL; + } + } + scr_http_dynh_not_fn(what); + (void)self; + return NULL; +} + +/* The sockets/requests snapshots: name → array (fresh objects per read; + * the tests read lengths, membership, and key presence — empty buckets + * are OMITTED, Node deletes drained keys). */ +static ScrDyn *scr_http_dynh_agent_table(ScrHttpAgent *a, bool queued) { + ScrDyn *obj = scr_dyn_new_obj(); + for (ScrHttpAgentEnt *e = a->ents; e; e = e->next) { + if (e->queued != queued) continue; + ScrDyn *item = queued + ? scr_dyn_new_handle(e->client, SCR_DYNH_HTTP_CLIENT) + : e->client->sock ? scr_dyn_new_handle(e->client->sock, SCR_DYNH_NET_SOCKET) : NULL; + if (item == NULL) continue; + ScrDyn *arr = scr_dyn_obj_get(obj, e->name->data, e->name->len); /* borrowed */ + if (arr == NULL || arr->kind != SCR_DYN_ARR) { + ScrDyn *fresh = scr_dyn_new_arr(); + scr_dyn_arr_push(fresh, item); /* moves */ + scr_dyn_obj_set(obj, e->name->data, e->name->len, fresh); /* moves */ + } else { + scr_dyn_arr_push(arr, item); /* moves */ + } + } + return obj; +} + +static ScrDyn *scr_http_dynh_agent_get(void *h, const char *key, size_t key_len) { + ScrHttpAgent *a = (ScrHttpAgent *)h; + (void)key_len; + if (strcmp(key, "maxSockets") == 0) return scr_dyn_new_num(a->max_sockets); + if (strcmp(key, "maxFreeSockets") == 0) return scr_dyn_new_num(a->max_free); + if (strcmp(key, "keepAlive") == 0) return scr_dyn_new_bool(a->keep_alive); + if (strcmp(key, "keepAliveMsecs") == 0) return scr_dyn_new_num(a->ka_msecs); + if (strcmp(key, "defaultPort") == 0) return scr_dyn_new_num(a->default_port); + if (strcmp(key, "protocol") == 0) { + ScrStr *s = scr_str_new(a->secure ? "https:" : "http:", a->secure ? 6 : 5); + ScrDyn *d = scr_dyn_new_str(s); + scr_str_release(s); + return d; + } + if (strcmp(key, "sockets") == 0) return scr_http_dynh_agent_table(a, false); + if (strcmp(key, "requests") == 0) return scr_http_dynh_agent_table(a, true); + if (strcmp(key, "freeSockets") == 0) { + /* honestly EMPTY: the runtime pools nothing (keepAlive fences) */ + return scr_dyn_new_obj(); + } + if (strcmp(key, "totalSocketCount") == 0) { + size_t n = 0; + for (ScrHttpAgentEnt *e = a->ents; e; e = e->next) { + if (!e->queued) n++; + } + return scr_dyn_new_num((double)n); + } + { + static const char *const known[] = { "options", "maxTotalSockets", "scheduling", NULL }; + if (scr_http_dynh_in(key, known)) { + scr_http_dynh_unsupported("Agent", key, NULL); + return NULL; + } + } + return NULL; +} + +static bool scr_http_dynh_agent_set(void *h, const char *key, size_t key_len, const ScrDyn *value) { + ScrHttpAgent *a = (ScrHttpAgent *)h; + (void)key_len; + if (strcmp(key, "defaultPort") == 0 || strcmp(key, "maxSockets") == 0 || + strcmp(key, "maxFreeSockets") == 0 || strcmp(key, "keepAliveMsecs") == 0) { + if (value->kind != SCR_DYN_NUM) { + scr_dyn_arg_type_fail(key, "of type number", value); + return true; /* handled: the exception is pending */ + } + if (strcmp(key, "defaultPort") == 0) a->default_port = value->v.num; + else if (strcmp(key, "maxSockets") == 0) a->max_sockets = value->v.num; + else if (strcmp(key, "maxFreeSockets") == 0) a->max_free = value->v.num; + else a->ka_msecs = value->v.num; + return true; + } + return false; /* createSocket/createConnection installs: the loud fence */ +} + +static const ScrDynHandleOps scr_http_dynh_agent_ops = { + "Agent", + &scr_http_agent_retain_v, + &scr_http_agent_release_v, + &scr_http_dynh_agent_invoke, + &scr_http_dynh_agent_get, + &scr_http_dynh_agent_set, + NULL, +}; + void scr_http_dyn_install(void) { scr_dyn_handle_install(SCR_DYNH_HTTP_REQ, &scr_http_dynh_req_ops); scr_dyn_handle_install(SCR_DYNH_HTTP_RES, &scr_http_dynh_res_ops); + scr_dyn_handle_install(SCR_DYNH_HTTP_CLIENT, &scr_http_dynh_client_ops); + scr_dyn_handle_install(SCR_DYNH_HTTP_AGENT, &scr_http_dynh_agent_ops); scr_net_set_dynh_http_on(&scr_http_dynh_server_on); } diff --git a/packages/runtime/src/scr_http2.c b/packages/runtime/src/scr_http2.c index 1140afffb..7673a9b3c 100644 --- a/packages/runtime/src/scr_http2.c +++ b/packages/runtime/src/scr_http2.c @@ -2192,6 +2192,7 @@ static void scr_h2_compat_dispatch(ScrH2Session *s, ScrH2Stream *st, const ScrH2 scr_http_h2_req_header(req, name, value); } ScrHttpRes *res = scr_http_h2_res_new(s->sock, st); + scr_http_res_set_req(res, req); /* res.req — the compat pair's backref */ st->compat_req = scr_http_req_retain(req); st->compat_res = scr_http_res_retain(res); /* the req owns the READ side: the stream's own 'end' bookkeeping is diff --git a/packages/runtime/src/scr_json.c b/packages/runtime/src/scr_json.c index 009df1822..c7b4e7065 100644 --- a/packages/runtime/src/scr_json.c +++ b/packages/runtime/src/scr_json.c @@ -1073,6 +1073,27 @@ ScrStr *scr_dyn_string_coerce(const ScrDyn *d) { * ignore silently — the loud choice, SEMANTICS.md). Receiver, key, and * value are all BORROWED (the member retains the value in). */ static const char *scr_dyn_kind_name(const ScrDyn *d); +/* `key in v` with a RUNTIME key (the compile-time dynHasKey fold, per + * value): OBJ answers own-member presence, ARR answers 'length' or a + * valid dense index, every other kind false (tsc admits `in` only on + * object-typed operands). Borrows both; never throws. */ +bool scr_dyn_has_key(const ScrDyn *v, const ScrStr *key) { + if (v->kind == SCR_DYN_OBJ) return scr_dyn_obj_get(v, key->data, key->len) != NULL; + if (v->kind == SCR_DYN_ARR) { + if (key->len == 6 && memcmp(key->data, "length", 6) == 0) return true; + if (key->len == 0 || key->len > 15) return false; + size_t idx = 0; + for (size_t i = 0; i < key->len; i++) { + char c = key->data[i]; + if (c < '0' || c > '9') return false; + if (i > 0 && idx == 0) return false; /* a leading zero is no canonical index */ + idx = idx * 10 + (size_t)(c - '0'); + } + return idx < v->v.arr.len; + } + return false; +} + void scr_dyn_key_set(ScrDyn *recv, ScrStr *key, ScrDyn *value) { if (recv->kind == SCR_DYN_OBJ) { scr_dyn_obj_set(recv, key->data, key->len, scr_dyn_retain(value)); diff --git a/packages/runtime/src/scr_net.c b/packages/runtime/src/scr_net.c index 387348a38..b08b66d10 100644 --- a/packages/runtime/src/scr_net.c +++ b/packages/runtime/src/scr_net.c @@ -463,12 +463,34 @@ struct ScrNetSocket { * dial_ips[dial_i]; the LAST failure's message surfaces (Node's * autoSelectFamily aggregate is the documented divergence) */ bool lookup_wait; + /* a deferred dial (the http agent's maxSockets queue): registered and + * "connecting" (writes buffer), the actual dial starts on + * scr_net_sock_dial_start */ + bool dial_deferred; ScrStr **dial_ips; size_t dial_n, dial_i; /* the in-flight dial chain's LAST failure — surfaced (moved into * pending_err) only when the list exhausts; a later success drops it */ ScrStr *dial_err; ScrNetLs data_ls, end_ls, close_ls, err_ls, conn_ls, timeout_ls, readable_ls; + /* Flow control (pause()/resume()): user_paused holds reads OFF even + * with consumers (kernel/TCP backpressure is the buffer — the honest + * pause); flowing counts as a consumer even with no 'data' listener + * (resume()'s flow-and-discard drain, how Node reaches 'end' on an + * unconsumed stream). resume() clears user_paused and sets flowing. */ + bool user_paused; + bool flowing; + /* destroySoon(): end() now, destroy once the FIN actually went out */ + bool destroy_on_finish; + /* socket.bytesWritten: every byte accepted by the write paths (user + * writes, protocol heads, pipe deliveries — plaintext on TLS sockets) */ + size_t bytes_written; + /* end(callback)/write(chunk, callback): 'finish'-shaped callbacks fire + * from the SWEEP (never the registering stack — Node defers them too). + * finish_ls waits for the FIN (wr_done); wcb_ls waits for the write + * buffer to drain (this surface's flush moment). */ + ScrNetLs finish_ls, wcb_ls; + bool finish_pending; /* wr_done seen: fire finish_ls at the next sweep */ /* a server whose 'connection' waits for this socket's handshake — the * TLS defer target (+1; the socket's own `server` backref stays the * ACCEPTING server, which may be the demux wrapper) */ @@ -563,6 +585,8 @@ void scr_net_sock_release(ScrNetSocket *s) { scr_net_ls_drop(&s->conn_ls); scr_net_ls_drop(&s->timeout_ls); scr_net_ls_drop(&s->readable_ls); + scr_net_ls_drop(&s->finish_ls); + scr_net_ls_drop(&s->wcb_ls); scr_str_release(s->pending_err); scr_str_release(s->remote_cache); for (size_t i = 0; i < s->dial_n; i++) scr_str_release(s->dial_ips[i]); @@ -750,8 +774,9 @@ static ScrNetSocket *scr_net_sock_new(void) { /* Consumer-driven read arming (the stdin discipline: never read a byte * nobody consumes — the socket buffer is the backpressure). */ static void scr_net_sock_update_read(ScrNetSocket *s) { - bool want = s->fd >= 0 && !s->connecting && !s->rd_eof && + bool want = s->fd >= 0 && !s->connecting && !s->rd_eof && !s->user_paused && (s->data_ls.n > 0 || s->pipe_dst != NULL || s->native_data != NULL || + s->flowing /* resume(): flow (and discard sans listeners) */ || s->readable_ls.n > 0 /* paused-mode consumer: bytes buffer */ || s->tops != NULL /* a transport is ALWAYS a consumer: the TLS * layer must process protocol frames (the @@ -803,6 +828,13 @@ static void scr_net_sock_maybe_finish_write(ScrNetSocket *s) { if (s->tops && s->t_est) s->tops->shutdown_write(s->tctx); shutdown(s->fd, SHUT_WR); s->wr_done = true; + if (s->finish_ls.n > 0) s->finish_pending = true; /* end(cb): the sweep fires it */ + if (s->destroy_on_finish) { + /* destroySoon(): the FIN is out — tear down now (Node's 'finish'-then- + * destroy). The close emits through the ordinary sweep. */ + scr_net_sock_close_fd(s); + return; + } if (s->rd_eof) scr_net_sock_close_fd(s); else scr_net_sock_update_read(s); /* the drain-to-EOF mode arms here */ } @@ -833,6 +865,7 @@ static void scr_net_sock_buffer(ScrNetSocket *s, const char *data, size_t len) { * SEMANTICS.md documents the bound). */ static void scr_net_sock_write_raw(ScrNetSocket *s, const char *data, size_t len) { if (s->fd < 0 || s->wr_ending || s->close_emitted || len == 0) return; + s->bytes_written += len; /* accepted bytes (Node counts buffered ones too) */ if (s->connecting || s->wlen > s->whead || (s->tops && !s->t_est)) { /* mid-connect and mid-handshake bytes buffer; the flush after * 'connect'/establishment sends them */ @@ -958,8 +991,8 @@ static void scr_net_sock_transport_pump(ScrNetSocket *s); static void scr_net_sock_read(ScrNetSocket *s) { if (s->tops && !s->t_est) return; /* readiness feeds the handshake pump instead */ for (;;) { - if (s->fd < 0) return; - bool flowing = s->data_ls.n > 0 || s->pipe_dst || s->native_data; + if (s->fd < 0 || s->user_paused) return; + bool flowing = s->data_ls.n > 0 || s->pipe_dst || s->native_data || s->flowing; /* buffered (peeked/unshifted) bytes first — they precede the * kernel's in the stream; a TLS engine drains them through its bio */ if (!s->tops && flowing && s->rlen > s->rhead) { @@ -1531,18 +1564,13 @@ ScrStr *scr_net_blocking_lookup(ScrStr *host /*borrowed*/) { return scr_str_new(ip, strlen(ip)); } -ScrNetSocket *scr_net_connect(double port, ScrStr *host /*borrowed, nullable*/, - ScrClosure *cb /*moves, nullable*/) { - ScrNetSocket *s = scr_net_sock_new(); - if (cb) scr_net_ls_add(&s->conn_ls, cb, NULL, true); - s->peer_port = (int)port; - const char *h = host && host->len > 0 ? host->data : "localhost"; - if (strcmp(h, "localhost") == 0) h = "127.0.0.1"; - snprintf(s->peer_ip, sizeof s->peer_ip, "%s", h); - if (!scr_net_poller_init()) { - fputs("scriptc: event poller init failed\n", stderr); - abort(); - } +/* The dial itself (peer_ip/peer_port already set): the sockaddr arms, + * socket(), and the nonblocking connect(). Shared by the immediate dial + * (scr_net_connect) and the DEFERRED one (the http agent's maxSockets + * queue — the socket exists, buffers writes, and dials when a slot + * frees). */ +static void scr_net_sock_dial_peer(ScrNetSocket *s) { + const char *h = s->peer_ip; struct sockaddr_in a4; struct sockaddr_in6 a6; struct sockaddr *sa = NULL; @@ -1568,8 +1596,8 @@ ScrNetSocket *scr_net_connect(double port, ScrStr *host /*borrowed, nullable*/, s->pending_err = scr_str_new(msg, strlen(msg)); s->had_error = true; s->emit_close = true; - scr_net_sock_register(s); - return s; + s->connecting = false; + return; } int fd = socket(sa->sa_family, SOCK_STREAM, 0); if (fd < 0) { @@ -1595,12 +1623,53 @@ ScrNetSocket *scr_net_connect(double port, ScrStr *host /*borrowed, nullable*/, s->peer_port); s->pending_err = scr_str_new(msg, strlen(msg)); s->had_error = true; + s->connecting = false; scr_net_sock_close_fd(s); } +} + +ScrNetSocket *scr_net_connect(double port, ScrStr *host /*borrowed, nullable*/, + ScrClosure *cb /*moves, nullable*/) { + ScrNetSocket *s = scr_net_sock_new(); + if (cb) scr_net_ls_add(&s->conn_ls, cb, NULL, true); + s->peer_port = (int)port; + const char *h = host && host->len > 0 ? host->data : "localhost"; + if (strcmp(h, "localhost") == 0) h = "127.0.0.1"; + snprintf(s->peer_ip, sizeof s->peer_ip, "%s", h); + if (!scr_net_poller_init()) { + fputs("scriptc: event poller init failed\n", stderr); + abort(); + } + scr_net_sock_dial_peer(s); scr_net_sock_register(s); return s; } +/* A dial the caller starts LATER (the http agent's maxSockets queue): + * the socket exists, registers, and buffers writes as "connecting" — + * scr_net_sock_dial_start runs the actual dial when a slot frees. */ +ScrNetSocket *scr_net_connect_deferred(double port, ScrStr *host /*borrowed, nullable*/) { + ScrNetSocket *s = scr_net_sock_new(); + s->peer_port = (int)port; + const char *h = host && host->len > 0 ? host->data : "localhost"; + if (strcmp(h, "localhost") == 0) h = "127.0.0.1"; + snprintf(s->peer_ip, sizeof s->peer_ip, "%s", h); + if (!scr_net_poller_init()) { + fputs("scriptc: event poller init failed\n", stderr); + abort(); + } + s->connecting = true; /* logically connecting while the dial waits */ + s->dial_deferred = true; + scr_net_sock_register(s); + return s; +} + +void scr_net_sock_dial_start(ScrNetSocket *s) { + if (!s->dial_deferred || s->close_emitted || s->emit_close || s->fd >= 0) return; + s->dial_deferred = false; + scr_net_sock_dial_peer(s); +} + /* ── the caller-lookup dial (net.connect with a lookup option) ───────── * * portless's createLoopbackConnection: connect({ host, port, @@ -1779,6 +1848,71 @@ void scr_net_sock_end_bytes(ScrNetSocket *s, ScrBytes *data /*borrowed*/) { scr_net_sock_end(s); } +/* socket.pause(): reads stay off — the kernel buffer (and TCP flow + * control) holds arrived bytes, the honest pause. Delivery restarts on + * resume(), always from the poller/sweep, never this stack. */ +void scr_net_sock_pause(ScrNetSocket *s) { + s->user_paused = true; + scr_net_sock_update_read(s); +} + +/* socket.resume(): flowing mode — a consumer even with no 'data' + * listener (arrived bytes discard, so 'end' can be reached, Node's + * resumed-but-unconsumed stream). Buffered bytes deliver from the next + * sweep (flags_pending sees the flowing consumer), not this stack. */ +void scr_net_sock_resume(ScrNetSocket *s) { + s->user_paused = false; + s->flowing = true; + scr_net_sock_update_read(s); +} + +/* socket.setNoDelay(enable): TCP_NODELAY on the live fd (client dials + * already set it — Node's default there is off, a documented divergence + * in the dial path; this call makes the state explicit either way). */ +void scr_net_sock_set_nodelay(ScrNetSocket *s, bool enable) { + if (s->fd < 0) return; + int v = enable ? 1 : 0; + setsockopt(s->fd, IPPROTO_TCP, TCP_NODELAY, &v, sizeof v); +} + +/* socket.destroySoon(): end the write half now, destroy once the FIN is + * actually out (buffered bytes flush first — Node's 'finish'-then-destroy). */ +void scr_net_sock_destroy_soon(ScrNetSocket *s) { + if (s->fd < 0) return; + if (s->wr_done) { + scr_net_sock_close_fd(s); + return; + } + s->destroy_on_finish = true; + s->wr_ending = true; + scr_net_sock_maybe_finish_write(s); +} + +/* end(callback): fires once the FIN went out ('finish'), from the sweep. */ +void scr_net_sock_on_finish(ScrNetSocket *s, ScrClosure *cb /*moves*/) { + if (s->close_emitted) { + scr_closure_release(cb); + return; + } + scr_net_ls_add(&s->finish_ls, cb, NULL, true); + if (s->wr_done) s->finish_pending = true; /* already finished: next sweep */ +} + +/* write(chunk, callback): fires when the write buffer drains (this + * surface's flush moment), from the sweep. */ +void scr_net_sock_on_write_flush(ScrNetSocket *s, ScrClosure *cb /*moves*/) { + if (s->close_emitted) { + scr_closure_release(cb); + return; + } + scr_net_ls_add(&s->wcb_ls, cb, NULL, true); +} + +double scr_net_sock_bytes_written(ScrNetSocket *s) { return (double)s->bytes_written; } + +/* socket.readable: true until the read half is done (peer FIN / destroy). */ +bool scr_net_sock_readable(ScrNetSocket *s) { return s->fd >= 0 && !s->rd_eof; } + /* socket.setTimeout(ms): ms > 0 arms the idle timer NOW (Node starts the * clock at the call — connecting time counts), ms <= 0 disables. */ void scr_net_sock_set_timeout(ScrNetSocket *s, double ms) { @@ -2239,10 +2373,13 @@ static bool scr_net_flags_pending(void) { if (scr_net_proto_pending()) return true; for (ScrNetSocket *s = scr_net_socks; s; s = s->next) { if (s->pending_err || (s->emit_close && !s->close_emitted)) return true; + if (s->finish_pending && s->finish_ls.n > 0) return true; + if (s->wcb_ls.n > 0 && (s->wlen == s->whead || s->fd < 0)) return true; /* bytes already out of the kernel with a consumer waiting — decrypted * plaintext inside a transport, or unshifted bytes in the receive * buffer: the poller can't re-signal them, the sweep must deliver */ - bool consumers = s->data_ls.n > 0 || s->pipe_dst || s->native_data; + bool consumers = (s->data_ls.n > 0 || s->pipe_dst || s->native_data || s->flowing) && + !s->user_paused; if (consumers && s->fd >= 0) { if (s->tops && s->t_est && s->tops->pending && s->tops->pending(s->tctx)) return true; if (!s->tops && s->rlen > s->rhead) return true; @@ -2281,7 +2418,8 @@ static void scr_net_sweep(void) { while (sock) { ScrNetSocket *next = sock->next; /* may unregister below */ scr_net_sock_retain(sock); /* callbacks may drop every other ref */ - if (sock->fd >= 0 && (sock->data_ls.n > 0 || sock->pipe_dst || sock->native_data) && + if (sock->fd >= 0 && !sock->user_paused && + (sock->data_ls.n > 0 || sock->pipe_dst || sock->native_data || sock->flowing) && ((sock->tops && sock->t_est && sock->tops->pending && sock->tops->pending(sock->tctx)) || (!sock->tops && sock->rlen > sock->rhead))) { /* deliver bytes the kernel can't re-signal (see flags_pending) */ @@ -2291,6 +2429,26 @@ static void scr_net_sweep(void) { return; } } + if (sock->wcb_ls.n > 0 && (sock->wlen == sock->whead || sock->fd < 0)) { + /* write(chunk, cb): the buffer drained (this surface's flush + * moment) — the callbacks fire off the sweep, never the writing + * stack. A dead socket fires them too (the buffered bytes are + * gone either way; Node errors them — documented divergence, + * the no-backpressure stance). */ + scr_net_fire0_this(&sock->wcb_ls, sock, SCR_DYNH_NET_SOCKET); + if (scr_exc_pending()) { + scr_net_sock_release(sock); + return; + } + } + if (sock->finish_pending) { + sock->finish_pending = false; + scr_net_fire0_this(&sock->finish_ls, sock, SCR_DYNH_NET_SOCKET); + if (scr_exc_pending()) { + scr_net_sock_release(sock); + return; + } + } if (sock->pending_err) { ScrStr *msg = sock->pending_err; sock->pending_err = NULL; @@ -2342,6 +2500,8 @@ static void scr_net_sweep(void) { scr_net_ls_drop(&sock->conn_ls); scr_net_ls_drop(&sock->timeout_ls); scr_net_ls_drop(&sock->readable_ls); + scr_net_ls_drop(&sock->finish_ls); + scr_net_ls_drop(&sock->wcb_ls); if (sock->pipe_dst) { scr_net_sock_release(sock->pipe_dst); sock->pipe_dst = NULL; @@ -2606,6 +2766,9 @@ static ScrDyn *scr_net_dynh_sock_invoke(void *h, ScrDyn *self, const char *metho scr_net_sock_on_timeout(s, scr_dyn_listener_closure0(cb), once); } else if (scr_net_dynh_name_is(name, "readable")) { scr_net_sock_on_readable(s, scr_dyn_listener_closure0(cb), once); + } else if (scr_net_dynh_name_is(name, "finish")) { + /* fires once when the FIN goes out — once either way */ + scr_net_sock_on_finish(s, scr_dyn_listener_closure0(cb)); } else if (scr_net_dynh_name_is(name, "drain")) { /* This surface never backpressures (write answers true) — an * accepted, never-fired registration is the consistent answer. */ @@ -2631,28 +2794,51 @@ static ScrDyn *scr_net_dynh_sock_invoke(void *h, ScrDyn *self, const char *metho scr_dyn_arg_type_fail("chunk", "of type string or an instance of Buffer or Uint8Array", chunk); return NULL; } - if (argc > 1 && args[1]->kind == SCR_DYN_FUNC) { - scr_net_dynh_unsupported("write", "write(chunk, callback) is not modeled"); - return NULL; - } if (chunk->kind == SCR_DYN_STR) scr_net_sock_write_str(s, chunk->v.str); else scr_net_sock_write_bytes(s, chunk->v.bytes); + if (argc > 1 && args[1]->kind == SCR_DYN_FUNC) { + /* write(chunk, cb): fires when the buffer drains (sweep-deferred) */ + scr_net_sock_on_write_flush(s, scr_dyn_listener_closure0(args[1])); + } return scr_dyn_new_bool(true); /* backpressure is not modeled (SEMANTICS.md) */ } if (strcmp(method, "end") == 0) { const ScrDyn *chunk = argc > 0 ? args[0] : scr_dyn_undefined(); + const ScrDyn *cb = NULL; + if (chunk->kind == SCR_DYN_FUNC) { /* end(callback) */ + cb = chunk; + chunk = scr_dyn_undefined(); + } else if (argc > 1 && args[1]->kind == SCR_DYN_FUNC) { + cb = args[1]; /* end(chunk, callback) */ + } + if (cb != NULL) scr_net_sock_on_finish(s, scr_dyn_listener_closure0(cb)); if (chunk->kind == SCR_DYN_STR) scr_net_sock_end_str(s, chunk->v.str); else if (chunk->kind == SCR_DYN_BYTES) scr_net_sock_end_bytes(s, chunk->v.bytes); else if (chunk->kind == SCR_DYN_UNDEF || argc == 0) scr_net_sock_end(s); - else if (chunk->kind == SCR_DYN_FUNC) { - scr_net_dynh_unsupported("end", "end(callback) is not modeled"); - return NULL; - } else { + else { scr_dyn_arg_type_fail("chunk", "of type string or an instance of Buffer or Uint8Array", chunk); return NULL; } return scr_dyn_retain(self); } + if (strcmp(method, "pause") == 0) { + scr_net_sock_pause(s); + return scr_dyn_retain(self); + } + if (strcmp(method, "resume") == 0) { + scr_net_sock_resume(s); + return scr_dyn_retain(self); + } + if (strcmp(method, "setNoDelay") == 0) { + /* setNoDelay([enable]) — missing/undefined means true, Node */ + bool enable = argc == 0 || args[0]->kind == SCR_DYN_UNDEF || scr_dyn_truthy(args[0]); + scr_net_sock_set_nodelay(s, enable); + return scr_dyn_retain(self); + } + if (strcmp(method, "destroySoon") == 0) { + scr_net_sock_destroy_soon(s); + return scr_dyn_retain(scr_dyn_undefined()); + } if (strcmp(method, "destroy") == 0) { if (argc > 0 && args[0]->kind != SCR_DYN_UNDEF) { scr_net_dynh_unsupported("destroy", "destroy(error) carries a payload this surface does not model"); @@ -2714,10 +2900,10 @@ static ScrDyn *scr_net_dynh_sock_invoke(void *h, ScrDyn *self, const char *metho if (scr_net_dynh_tls->invoke(s, method, args, argc, &out)) return out; } { - static const char *const known[] = { "connect", "setNoDelay", "setKeepAlive", - "address", "pause", "resume", "ref", "unref", "cork", "uncork", "read", "unpipe", + static const char *const known[] = { "connect", "setKeepAlive", + "address", "ref", "unref", "cork", "uncork", "read", "unpipe", "off", "removeListener", "removeAllListeners", "emit", "prependListener", - "prependOnceListener", "listenerCount", "listeners", "resetAndDestroy", "destroySoon", NULL }; + "prependOnceListener", "listenerCount", "listeners", "resetAndDestroy", NULL }; for (size_t i = 0; known[i]; i++) { if (strcmp(method, known[i]) == 0) { scr_net_dynh_unsupported(method, NULL); @@ -2750,14 +2936,21 @@ static ScrDyn *scr_net_dynh_sock_get(void *h, const char *key, size_t key_len) { * undefined — the static lane's false→undefined mapping. */ return scr_net_sock_encrypted(s) ? scr_dyn_new_bool(true) : NULL; } + if (strcmp(key, "bytesWritten") == 0) return scr_dyn_new_num(scr_net_sock_bytes_written(s)); + if (strcmp(key, "readable") == 0) return scr_dyn_new_bool(scr_net_sock_readable(s)); + if (strcmp(key, "writableHighWaterMark") == 0 || strcmp(key, "readableHighWaterMark") == 0) { + /* Node's default stream highWaterMark — a constant here (backpressure + * is not modeled; SEMANTICS.md) */ + return scr_dyn_new_num(16384); + } if (scr_net_dynh_tls != NULL) { ScrDyn *out = NULL; if (scr_net_dynh_tls->get(s, key, &out)) return out; } { static const char *const known[] = { "remotePort", "remoteFamily", "localAddress", - "localPort", "localFamily", "bytesRead", "bytesWritten", "connecting", "pending", - "readyState", "bufferSize", "timeout", "readable", "closed", "errored", NULL }; + "localPort", "localFamily", "bytesRead", "connecting", "pending", + "readyState", "bufferSize", "timeout", "closed", "errored", NULL }; for (size_t i = 0; known[i]; i++) { if (strcmp(key, known[i]) == 0) { scr_net_dynh_unsupported(key, NULL); diff --git a/packages/runtime/src/scr_runtime.h b/packages/runtime/src/scr_runtime.h index f5bc949cf..f482b0e4c 100644 --- a/packages/runtime/src/scr_runtime.h +++ b/packages/runtime/src/scr_runtime.h @@ -2591,6 +2591,8 @@ typedef enum { SCR_DYNH_NET_SERVER, /* ScrNetServer — net.Server (http.Server rides the same handle) */ SCR_DYNH_H2_SESSION, /* ScrH2Session — Http2Session (client & server) */ SCR_DYNH_H2_STREAM, /* ScrH2Stream — Http2Stream (client & server) */ + SCR_DYNH_HTTP_CLIENT, /* ScrHttpClientReq — http.ClientRequest */ + SCR_DYNH_HTTP_AGENT, /* ScrHttpAgent — http.Agent / https.Agent */ SCR_DYNH_COUNT, } ScrDynHandleTag; @@ -2712,6 +2714,9 @@ void scr_dyn_obj_set(ScrDyn *obj, const char *key, size_t key_len, ScrDyn *value * non-object kinds throw Node's catchable TypeErrors (strict-mode * wording). All three operands BORROWED (the value is retained in). */ void scr_dyn_key_set(ScrDyn *recv, ScrStr *key, ScrDyn *value); +/* `key in v` with a runtime key — the dynHasKey fold per value (OBJ own + * members, ARR length/valid indices, false elsewhere). Never throws. */ +bool scr_dyn_has_key(const ScrDyn *v, const ScrStr *key); /* Bare `typeof v` on a dyn value: the DOM kind's JS answer (+1 string; * null answers "object"). Never throws. */ ScrStr *scr_dyn_typeof(const ScrDyn *d); @@ -4411,6 +4416,22 @@ void scr_net_sock_end_bytes(ScrNetSocket *s, ScrBytes *data /*borrowed*/); void scr_net_sock_write_dynv(ScrNetSocket *s, const ScrDyn *d /*borrowed*/); void scr_net_sock_end_dynv(ScrNetSocket *s, const ScrDyn *d /*borrowed*/); void scr_net_sock_destroy(ScrNetSocket *s); +/* Flow control (pause/resume — see the struct's flag comments), the + * FIN-flushed destroy (destroySoon), TCP_NODELAY, the deferred write/ + * finish callbacks (write(chunk, cb) / end(cb) — sweep-fired), and the + * counters/flags the compat surface reads. */ +void scr_net_sock_pause(ScrNetSocket *s); +void scr_net_sock_resume(ScrNetSocket *s); +void scr_net_sock_set_nodelay(ScrNetSocket *s, bool enable); +void scr_net_sock_destroy_soon(ScrNetSocket *s); +void scr_net_sock_on_finish(ScrNetSocket *s, ScrClosure *cb /*moves*/); +void scr_net_sock_on_write_flush(ScrNetSocket *s, ScrClosure *cb /*moves*/); +double scr_net_sock_bytes_written(ScrNetSocket *s); +bool scr_net_sock_readable(ScrNetSocket *s); +/* The deferred dial (the http agent's maxSockets queue): the socket + * registers "connecting" and buffers writes; dial_start runs the dial. */ +ScrNetSocket *scr_net_connect_deferred(double port, ScrStr *host /*borrowed, nullable*/); /* +1 */ +void scr_net_sock_dial_start(ScrNetSocket *s); void scr_net_sock_set_timeout(ScrNetSocket *s, double ms); void scr_net_sock_on_timeout(ScrNetSocket *s, ScrClosure *cb /*moves*/, bool once); ScrStr *scr_net_sock_remote_address(ScrNetSocket *s); /* +1 or NULL (undefined arm) */ @@ -4765,6 +4786,23 @@ void scr_http_upgrade_thunk3(ScrClosure *cb, ScrHttpReq *req, ScrNetSocket *sock double scr_http_req_status(ScrHttpReq *r); /* < 0 = the undefined arm (server request) */ ScrNetSocket *scr_http_req_socket(ScrHttpReq *r); /* +1 */ void scr_http_req_resume(ScrHttpReq *r); +/* pause()/resume() hold and drain 'data'/'end' delivery (the parser keeps + * consuming; the drain rides the emit queue); setTimeout delegates to the + * socket's idle timer; the flags back req.destroyed/req.readable. */ +void scr_http_req_pause(ScrHttpReq *r); +void scr_http_req_set_timeout(ScrHttpReq *r, double ms, ScrClosure *cb /*moves, nullable*/); +bool scr_http_req_destroyed_flag(ScrHttpReq *r); +bool scr_http_req_readable(ScrHttpReq *r); +/* flushHeaders/cork/uncork/writableCorked, the res.req backref, the + * socket-delegated setTimeout, and write(chunk, cb)'s deferred callback. */ +void scr_http_res_flush_headers(ScrHttpRes *r); +void scr_http_res_cork(ScrHttpRes *r); +void scr_http_res_uncork(ScrHttpRes *r); +double scr_http_res_writable_corked(ScrHttpRes *r); +bool scr_http_res_destroyed_flag(ScrHttpRes *r); +void scr_http_res_set_req(ScrHttpRes *r, ScrHttpReq *req /*borrowed, nullable*/); +void scr_http_res_set_timeout(ScrHttpRes *r, double ms, ScrClosure *cb /*moves, nullable*/); +void scr_http_res_on_write_flush(ScrHttpRes *r, ScrClosure *cb /*moves*/); /* req.setEncoding(enc) — the socket twin's contract; may throw. */ void scr_http_req_set_encoding(ScrHttpReq *r, ScrStr *enc /*borrowed*/); void scr_http_req_destroy(ScrHttpReq *r); @@ -4829,6 +4867,30 @@ ScrHttpClientReq *scr_http_request_ex(ScrStr *host /*borrowed*/, double port, ScrHttpClientReq *scr_http_request_url(ScrStr *url /*borrowed*/, ScrStr *method /*borrowed*/, bool auto_end, ScrClosure *cb /*moves, nullable*/, ScrHttpRespFn fn); /* +1 */ +/* ── the http Agent (new http.Agent(opts) — option surface, getName, and + * the maxSockets queue over one-dial-per-request connections; keep-alive + * POOLING is not modeled: keepAlive: true fences at construction). + * agent_new answers the SCR_DYNH_HTTP_AGENT handle (+1 dyn) or throws. + * max_sockets/max_free arrive < 0 for "unset" (Infinity / 256); + * timeout_ms < 0 = no idle timer. request_agent_ex threads the agent dyn + * (undefined/null = the default path; false = one-shot with Connection: + * close; an Agent handle = queue accounting); port < 0 means "no port + * option" — the agent's (settable) defaultPort, then the scheme's. */ +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: the option-merge default */); /* +1 */ +ScrHttpClientReq *scr_http_request_agent_ex(ScrStr *host /*borrowed*/, double port, + ScrStr *path /*borrowed*/, ScrStr *method /*borrowed*/, + double timeout_ms, ScrArr *header_pairs /*borrowed*/, + bool auto_end, const ScrDyn *agent /*borrowed*/, + ScrClosure *cb /*moves, nullable*/, + ScrHttpRespFn fn, int default_port, + void (*wrap)(ScrNetSocket *, void *), void *wrap_ctx); /* +1 */ +ScrHttpClientReq *scr_http_request_agent(ScrStr *host /*borrowed*/, double port, + ScrStr *path /*borrowed*/, ScrStr *method /*borrowed*/, + double timeout_ms, ScrArr *header_pairs /*borrowed*/, + bool auto_end, const ScrDyn *agent /*borrowed*/, + ScrClosure *cb /*moves, nullable*/, ScrHttpRespFn fn); /* +1 */ #ifdef SCR_RC_AUDIT long scr_http_live_count(void); #endif @@ -4979,6 +5041,14 @@ ScrHttpClientReq *scr_https_request(ScrStr *host /*borrowed*/, double port, bool auto_end, bool reject_unauthorized, const char *ca /*borrowed, len 0 = none*/, size_t ca_len, ScrClosure *cb /*moves, nullable*/, ScrHttpRespFn fn); /* +1 */ +/* The agent-threaded twin (the scr_http_request_agent story over TLS). */ +ScrHttpClientReq *scr_https_request_agent(ScrStr *host /*borrowed*/, double port, + ScrStr *path /*borrowed*/, ScrStr *method /*borrowed*/, + double timeout_ms, ScrArr *header_pairs /*borrowed*/, + bool auto_end, bool reject_unauthorized, + const char *ca /*borrowed, len 0 = none*/, size_t ca_len, + const struct ScrDyn *agent /*borrowed*/, + ScrClosure *cb /*moves, nullable*/, ScrHttpRespFn fn); /* +1 */ /* The fetch unit's https leg (defined in scr_tls.c): a client transport * context (SNI/verify against the URL hostname — the DIALED address may * be a resolved IP) plus the wrap hook scr_http_request_ex installs on diff --git a/packages/runtime/src/scr_tls.c b/packages/runtime/src/scr_tls.c index 14c4e344a..e96a5c8e0 100644 --- a/packages/runtime/src/scr_tls.c +++ b/packages/runtime/src/scr_tls.c @@ -982,6 +982,18 @@ ScrHttpClientReq *scr_https_request(ScrStr *host /*borrowed*/, double port, fn, 443, &scr_tls_cli_wrap, cli); } +ScrHttpClientReq *scr_https_request_agent(ScrStr *host /*borrowed*/, double port, + ScrStr *path /*borrowed*/, ScrStr *method /*borrowed*/, + double timeout_ms, ScrArr *header_pairs /*borrowed*/, + bool auto_end, bool reject_unauthorized, + const char *ca /*borrowed, len 0 = none*/, size_t ca_len, + const ScrDyn *agent /*borrowed*/, + ScrClosure *cb /*moves, nullable*/, ScrHttpRespFn fn) { + ScrTlsCli *cli = scr_tls_cli_new(host, reject_unauthorized, ca, ca_len); + return scr_http_request_agent_ex(host, port, path, method, timeout_ms, header_pairs, auto_end, + agent, cb, fn, 443, &scr_tls_cli_wrap, cli); +} + /* ── RUNTIME options records (the divergence-66 stance) ──────────────── * A non-literal options value — the checked-dynamic JS lane's DOM record * — reads its members at RUNTIME. Members the literal path lowers take diff --git a/tests/corpus/2573-http-dyn-res-compat.cjs b/tests/corpus/2573-http-dyn-res-compat.cjs new file mode 100644 index 000000000..0b84ad885 --- /dev/null +++ b/tests/corpus/2573-http-dyn-res-compat.cjs @@ -0,0 +1,87 @@ +// The ServerResponse compat members on CHECKED-DYNAMIC handles (the +// untyped-wrapper listener boxes req/res into the DOM): setHeader +// chaining + getHeaders snapshots, flushHeaders, the raw-array writeHead +// form (per-name override of setHeader state, other names survive), +// cork/uncork with the writableCorked counter (corked writes coalesce and +// flush on the last uncork), write-with-encoding ('hex' decodes the +// string chunk), the deferred write/end callbacks, res.req identity, and +// the destroyed flags (false at work time, true inside 'close'). +// Strict ping-pong: one exchange at a time, every line causally ordered. +'use strict'; +const http = require('http'); + +function wrap(fn) { + return function () { + return fn.apply(this, arguments); + }; +} + +const server = http.createServer(); +server.on('request', wrap(function (req, res) { + console.log('srv', req.url, 'req.destroyed:', req.destroyed, 'req.readable:', req.readable); + res.setHeader('a', '1'); + console.log('setHeader chains:', res.setHeader('b', '2') === res); + console.log('getHeaders:', JSON.stringify(res.getHeaders())); + if (req.url === '/raw') { + res.writeHead(200, ['test', '9', 'b', 'override']); + res.end('raw'); + return; + } + if (req.url === '/cork') { + res.writeHead(200); + console.log('corked0:', res.writableCorked); + res.cork(); + res.write('AA'); + res.cork(); + res.write('BB'); + console.log('corked2:', res.writableCorked); + res.uncork(); + res.uncork(); + console.log('corked-after:', res.writableCorked, 'finished:', res.writableFinished); + res.end('CC', wrap(function () { + console.log('end cb, finished:', res.writableFinished); + })); + return; + } + if (req.url === '/hex') { + res.writeHead(200); + res.write('414243', 'hex', wrap(function () { + console.log('write cb ran'); + })); + res.end(); + return; + } + console.log('res.req is req:', res.req === req); + res.flushHeaders(); + console.log('headersSent after flushHeaders:', res.headersSent); + res.on('close', wrap(function () { + console.log('res close, destroyed:', res.destroyed); + })); + res.end('done'); +})); + +function get(path, done) { + http.get({ port: server.address().port, path: path }, wrap(function (res) { + let body = ''; + res.on('data', wrap(function (c) { + body += c.toString('utf8'); + })); + res.on('end', wrap(function () { + console.log(path, '->', res.statusCode, JSON.stringify(body), + 'test:', res.headers.test, 'a:', res.headers.a, 'b:', res.headers.b); + done(); + })); + })); +} + +server.listen(0, function () { + get('/plain', function () { + get('/raw', function () { + get('/cork', function () { + get('/hex', function () { + server.close(); + }); + }); + }); + }); +}); diff --git a/tests/corpus/2574-http-dyn-req-pause.cjs b/tests/corpus/2574-http-dyn-req-pause.cjs new file mode 100644 index 000000000..6e9e98d14 --- /dev/null +++ b/tests/corpus/2574-http-dyn-req-pause.cjs @@ -0,0 +1,51 @@ +// IncomingMessage flow control on checked-dynamic handles: req.pause() +// holds 'data' delivery (the buffered bytes drain on resume(), off the +// resuming stack) AND defers 'end' until the drain; the client response +// reads readable/destroyed (true until 'end', destroyed only at +// teardown). The timings are causal, not raced: the pause happens on the +// first delivered chunk, the resume 30ms later, and the request's final +// chunk lands 60ms in — after the resume, so the body always reassembles +// complete. +'use strict'; +const http = require('http'); + +function wrap(fn) { + return function () { + return fn.apply(this, arguments); + }; +} + +const server = http.createServer(); +server.on('request', wrap(function (req, res) { + let got = ''; + let paused = false; + req.on('data', wrap(function (c) { + got += c.toString('utf8'); + if (!paused) { + paused = true; + console.log('pausing after first chunk:', got.length >= 4); + req.pause(); + setTimeout(wrap(function () { + console.log('resuming'); + req.resume(); + }), 30); + } + })); + req.on('end', wrap(function () { + console.log('server end, body:', JSON.stringify(got)); + res.end('ok'); + })); +})); + +server.listen(0, function () { + const r = http.request({ port: server.address().port, method: 'POST', path: '/' }, wrap(function (res) { + console.log('res.readable pre:', res.readable, 'destroyed pre:', res.destroyed); + res.resume(); + res.on('end', wrap(function () { + console.log('res.readable post:', res.readable); + server.close(); + })); + })); + r.write('abcd'); + setTimeout(function () { r.end('efghij'); }, 60); +}); diff --git a/tests/corpus/2575-net-dyn-socket-compat.cjs b/tests/corpus/2575-net-dyn-socket-compat.cjs new file mode 100644 index 000000000..50a3959af --- /dev/null +++ b/tests/corpus/2575-net-dyn-socket-compat.cjs @@ -0,0 +1,53 @@ +// The net.Socket compat members on checked-dynamic handles: setNoDelay +// chains, pause()/resume() return the socket (and genuinely gate reads), +// write(chunk, cb)/end(chunk, cb) fire their callbacks off the sweep (cb +// after the bytes entered the buffer / the FIN went out), bytesWritten +// counts accepted bytes, readable flips at the read half's end, +// destroySoon tears down after the FIN, and dyn Buffer at()/slice() +// dispatch (slice COPIES — the documented no-views stance — and keeps +// the Buffer flavor). One connection, causally ordered lines; the pause +// guard fires once no matter how the chunks arrive. +'use strict'; +const net = require('net'); + +function wrap(fn) { + return function () { + return fn.apply(this, arguments); + }; +} + +const server = net.createServer(wrap(function (sock) { + console.log('setNoDelay is sock:', sock.setNoDelay() === sock); + console.log('readable pre:', sock.readable); + sock.write('hello', wrap(function () { + console.log('write cb, bytesWritten:', sock.bytesWritten); + })); + sock.end('!', wrap(function () { + console.log('finish cb'); + })); +})); + +server.listen(0, wrap(function () { + const c = net.connect(server.address().port); + let got = ''; + let pausedOnce = false; + c.on('data', wrap(function (ch) { + got += ch.toString('utf8'); + if (!pausedOnce) { + pausedOnce = true; + console.log('pause returns c:', c.pause() === c); + setTimeout(wrap(function () { c.resume(); }), 20); + } + })); + c.on('end', wrap(function () { + console.log('client got:', JSON.stringify(got), 'readable:', c.readable); + const dyn = wrap(function (b) { return b; }); + const buf = dyn(Buffer.from('abc\n')); + console.log('at(-1):', buf.at(-1), 'slice:', buf.slice(0, 3).toString('utf8')); + c.destroySoon(); + })); + c.on('close', wrap(function () { + console.log('client closed, destroyed:', c.destroyed); + server.close(); + })); +})); diff --git a/tests/corpus/2576-http-agent-surface.cjs b/tests/corpus/2576-http-agent-surface.cjs new file mode 100644 index 000000000..dedff6d9c --- /dev/null +++ b/tests/corpus/2576-http-agent-surface.cjs @@ -0,0 +1,44 @@ +// The http.Agent surface: construction with options, getName's exact +// string shapes (host/port/localAddress/family/socketPath arms), the +// option properties (maxSockets Infinity default, maxFreeSockets, +// protocol, settable defaultPort), a request THROUGH the agent (the +// sockets/requests tables key by getName; freeSockets stays empty — no +// pooling exists), and destroy(). The default-port merge: an agent +// carrying defaultPort dials it when the request omits port. Causally +// ordered, one exchange at a time. +'use strict'; +const http = require('http'); + +const agent = new http.Agent(); +console.log('getName():', agent.getName()); +console.log('getName({}):', agent.getName({})); +console.log('getName(full):', agent.getName({ host: '0.0.0.0', port: 80, localAddress: '192.168.1.1' })); +console.log('getName(family4):', agent.getName({ family: 4 })); +console.log('getName(sockPath):', agent.getName({ socketPath: '/tmp/x.sock' })); +console.log('maxSockets:', agent.maxSockets, 'maxFreeSockets:', agent.maxFreeSockets); +console.log('protocol:', agent.protocol, 'keepAlive:', agent.keepAlive); + +const server = http.createServer(function (req, res) { + res.end('ok:' + req.url); +}); + +server.listen(0, function () { + const port = server.address().port; + agent.defaultPort = port; + console.log('defaultPort set:', agent.defaultPort === port); + const name = agent.getName({ port: port }); + // No port option: the agent's defaultPort dials (Node's option merge). + http.get({ host: 'localhost', agent: agent, path: '/merged' }, function (res) { + console.log('status:', res.statusCode); + console.log('sockets has name:', name in agent.sockets, 'len:', agent.sockets[name].length); + console.log('requests has name:', name in agent.requests); + console.log('freeSockets empty:', JSON.stringify(agent.freeSockets)); + let body = ''; + res.on('data', function (c) { body += c.toString('utf8'); }); + res.on('end', function () { + console.log('body:', body); + agent.destroy(); + server.close(); + }); + }); +}); diff --git a/tests/corpus/2577-http-agent-maxsockets.cjs b/tests/corpus/2577-http-agent-maxsockets.cjs new file mode 100644 index 000000000..e3cffae7c --- /dev/null +++ b/tests/corpus/2577-http-agent-maxsockets.cjs @@ -0,0 +1,38 @@ +// REAL maxSockets accounting through an Agent: with maxSockets 1 the +// second request queues (agent.requests[name] holds it) until the first +// exchange finishes, and the counters read exactly Node's at every +// causally-pinned point — sockets[name].length at response time, the +// queue length, and the drained tables afterwards. Socket REUSE is not +// pinned (this client dials one connection per request — the documented +// divergence); only the counters and the strict first-then-second +// ordering are. +'use strict'; +const http = require('http'); + +const agent = new http.Agent({ maxSockets: 1 }); +const server = http.createServer(function (req, res) { + res.end('ok:' + req.url); +}); + +server.listen(0, function () { + const port = server.address().port; + const name = agent.getName({ port: port }); + let done = 0; + const finish = function () { + if (++done === 2) { + console.log('after both: sockets?', name in agent.sockets, 'requests?', name in agent.requests); + agent.destroy(); + server.close(); + } + }; + http.get({ port: port, agent: agent, path: '/a' }, function (res) { + console.log('res1', res.statusCode, 'active:', agent.sockets[name].length, 'queued:', agent.requests[name].length); + res.resume(); + res.on('end', finish); + }); + http.get({ port: port, agent: agent, path: '/b' }, function (res2) { + console.log('res2', res2.statusCode, 'still queued:', name in agent.requests, 'active:', agent.sockets[name].length); + res2.resume(); + res2.on('end', finish); + }); +}); diff --git a/tests/diagnostics/options-record.ts b/tests/diagnostics/options-record.ts index 89d133a36..cc809493a 100644 --- a/tests/diagnostics/options-record.ts +++ b/tests/diagnostics/options-record.ts @@ -10,9 +10,10 @@ import * as fs from "node:fs"; import * as dns from "node:dns"; import * as readline from "node:readline"; -// A pooling Agent instance cannot be honored — pointed fence at the -// construction AND at the option. -const agent = new http.Agent({ keepAlive: true }); +// The Agent lowers to a checked-dynamic handle now — but only from a +// LITERAL options object; a runtime options value cannot be walked. +const agentOpts: http.AgentOptions = { maxSockets: 1 }; +const agent = new http.Agent(agentOpts); http.get({ port: 80, agent }, () => {}); // agent: false needs a LITERAL headers object to inject Connection: close