diff --git a/packages/compiler/ambient/scriptc-node-fallback.d.ts b/packages/compiler/ambient/scriptc-node-fallback.d.ts index e988f1dad..1d6db6315 100644 --- a/packages/compiler/ambient/scriptc-node-fallback.d.ts +++ b/packages/compiler/ambient/scriptc-node-fallback.d.ts @@ -690,6 +690,7 @@ interface RequestInit { headers?: Record; body?: string | Uint8Array | ReadableStream | null; duplex?: "half"; + redirect?: "follow" | "error" | "manual"; signal?: AbortSignal; } declare function fetch(input: string | URL, init?: RequestInit): Promise; diff --git a/packages/runtime/src/scr_fetch.c b/packages/runtime/src/scr_fetch.c index 854b846c1..44aefe646 100644 --- a/packages/runtime/src/scr_fetch.c +++ b/packages/runtime/src/scr_fetch.c @@ -100,15 +100,21 @@ typedef struct SfReader SfReader; typedef struct SfResponse SfResponse; typedef struct SfHeaders SfHeaders; typedef struct SfCollector SfCollector; +typedef struct SfReadRequest SfReadRequest; typedef struct SfPullWait SfPullWait; typedef struct SfStartWait SfStartWait; typedef struct SfCancelWait SfCancelWait; typedef struct SfChunk { - ScrBytes *bytes; + ScrDyn *value; struct SfChunk *next; } SfChunk; +struct SfReadRequest { + ScrPromise *promise; + SfReadRequest *next; +}; + typedef struct SfAbortListener { ScrDyn *fn; bool once; @@ -138,8 +144,10 @@ struct SfSignal { struct SfReader { size_t rc; - SfStream *stream; /* weak; the locked stream owns one reader ref */ - ScrPromise *pending; + size_t handle_refs; + SfStream *stream; /* weak; handle refs retain it while the lock is live */ + SfReadRequest *pending_head; + SfReadRequest *pending_tail; ScrPromise *closed; }; @@ -149,6 +157,12 @@ enum { SF_COLLECT_BYTES, }; +enum { + SF_REDIRECT_FOLLOW, + SF_REDIRECT_ERROR, + SF_REDIRECT_MANUAL, +}; + struct SfCollector { ScrPromise *promise; SfStream *held_stream; @@ -216,6 +230,7 @@ struct SfTransfer { ScrStr *method; /* current method, owned */ ScrArr *headers; /* user header pairs, owned */ ScrDyn *body; /* replayable body or source stream, owned */ + int redirect_mode; int hops; bool redirected; ScrHttpClientReq *client; /* owned */ @@ -646,23 +661,35 @@ static SfReader *sf_reader_retain(SfReader *r) { return r; } +static SfReadRequest *sf_reader_take_request(SfReader *r) { + SfReadRequest *request = r->pending_head; + if (!request) return NULL; + r->pending_head = request->next; + if (!r->pending_head) r->pending_tail = NULL; + request->next = NULL; + return request; +} + +static void sf_reader_reject_message_all(SfReader *r, const char *message) { + SfReadRequest *request; + while ((request = sf_reader_take_request(r)) != NULL) { + sf_type_error(message); + scr_promise_reject_pending(request->promise); + scr_promise_release(request->promise); + free(request); + } +} + static void sf_reader_release(SfReader *r) { if (!r || --r->rc > 0) return; - scr_promise_release(r->pending); + sf_reader_reject_message_all(r, "ReadableStream was released"); scr_promise_release(r->closed); free(r); } -static void *sf_reader_retain_v(void *p) { - return sf_reader_retain((SfReader *)p); -} -static void sf_reader_release_v(void *p) { - sf_reader_release((SfReader *)p); -} - static void sf_chunk_release(SfChunk *c) { if (!c) return; - scr_bytes_release(c->bytes); + scr_dyn_release(c->value); free(c); } @@ -680,13 +707,9 @@ static void sf_stream_release(SfStream *s) { sf_stream_drop_chunks(s); if (s->reader) { s->reader->stream = NULL; - if (s->reader->pending) { - sf_type_error("ReadableStream was released"); - scr_promise_reject_pending(s->reader->pending); - scr_promise_release(s->reader->pending); - s->reader->pending = NULL; - } + sf_reader_reject_message_all(s->reader, "ReadableStream was released"); sf_reader_release(s->reader); + s->reader = NULL; } scr_dyn_release(s->error); scr_dyn_release(s->pull_cb); @@ -701,6 +724,29 @@ static void sf_stream_release_v(void *p) { sf_stream_release((SfStream *)p); } +/* + * The stream owns its active reader so the lock survives loss of the + * user-visible reader box. Conversely, every reader HANDLE retains the + * stream while the lock is live. Keeping those two ownership classes + * distinct avoids a permanent stream↔reader cycle while making chained + * expressions such as `new ReadableStream(...).getReader()` safe. + */ +static void *sf_reader_handle_retain_v(void *p) { + SfReader *r = p; + sf_reader_retain(r); + r->handle_refs++; + if (r->stream) sf_stream_retain(r->stream); + return r; +} + +static void sf_reader_handle_release_v(void *p) { + SfReader *r = p; + SfStream *stream = r->stream; + if (r->handle_refs > 0) r->handle_refs--; + sf_reader_release(r); + sf_stream_release(stream); +} + static ScrBytes *sf_chunk_bytes(const ScrDyn *chunk) { if (chunk && chunk->kind == SCR_DYN_BYTES) { return scr_bytes_copy(chunk->v.bytes); @@ -714,30 +760,33 @@ static ScrBytes *sf_chunk_bytes(const ScrDyn *chunk) { return NULL; } -static ScrDyn *sf_read_result(bool done, ScrBytes *bytes) { +static ScrDyn *sf_read_result(bool done, ScrDyn *value) { ScrDyn *result = scr_dyn_new_obj(); scr_dyn_obj_set(result, "done", 4, scr_dyn_new_bool(done)); scr_dyn_obj_set(result, "value", 5, - bytes ? scr_dyn_new_bytes_copy(bytes) + value ? scr_dyn_retain(value) : scr_dyn_retain(scr_dyn_undefined())); return result; } -static void sf_reader_fulfill(SfReader *r, bool done, ScrBytes *bytes) { - if (!r || !r->pending) return; - ScrPromise *p = r->pending; - r->pending = NULL; - scr_promise_fulfill_ref(p, sf_read_result(done, bytes), +static void sf_reader_fulfill_one(SfReader *r, bool done, ScrDyn *value) { + if (!r) return; + SfReadRequest *request = sf_reader_take_request(r); + if (!request) return; + scr_promise_fulfill_ref(request->promise, sf_read_result(done, value), &scr_dyn_retain_v, &scr_dyn_release_v, NULL); - scr_promise_release(p); + scr_promise_release(request->promise); + free(request); } -static void sf_reader_reject(SfReader *r, ScrDyn *reason) { - if (!r || !r->pending) return; - ScrPromise *p = r->pending; - r->pending = NULL; - sf_reject_promise_reason(p, reason); - scr_promise_release(p); +static void sf_reader_reject_all(SfReader *r, ScrDyn *reason) { + if (!r) return; + SfReadRequest *request; + while ((request = sf_reader_take_request(r)) != NULL) { + sf_reject_promise_reason(request->promise, reason); + scr_promise_release(request->promise); + free(request); + } } static void sf_collector_append(SfCollector *c, const ScrBytes *bytes) { @@ -778,20 +827,20 @@ static void sf_collector_finish(SfStream *s) { SfCollector *c = s->collector; if (!c) return; ScrDyn *value = NULL; + ScrBytes *bytes = scr_bytes_new(SCR_BYTES_U8, (double)c->len); + if (c->len) memcpy(bytes->data, c->data, c->len); if (c->mode == SF_COLLECT_JSON) { - ScrStr *text = scr_str_new(c->data ? c->data : "", c->len); + ScrStr *text = scr_text_decode(bytes); value = scr_json_parse(text); scr_str_release(text); } else if (c->mode == SF_COLLECT_TEXT) { - ScrStr *text = scr_str_new(c->data ? c->data : "", c->len); + ScrStr *text = scr_text_decode(bytes); value = scr_dyn_new_str(text); scr_str_release(text); } else { - ScrBytes *bytes = scr_bytes_new(SCR_BYTES_U8, (double)c->len); - if (c->len) memcpy(bytes->data, c->data, c->len); value = scr_dyn_new_bytes_copy(bytes); - scr_bytes_release(bytes); } + scr_bytes_release(bytes); if (scr_exc_pending()) { scr_dyn_release(value); scr_promise_reject_pending(c->promise); @@ -829,7 +878,16 @@ static void sf_stream_drain(SfStream *s) { SfChunk *c = s->head; s->head = c->next; if (!s->head) s->tail = NULL; - sf_collector_append(s->collector, c->bytes); + ScrBytes *bytes = sf_chunk_bytes(c->value); + if (!bytes) { + sf_chunk_release(c); + scr_promise_reject_pending(s->collector->promise); + sf_collector_drop(s); + sf_stream_drop_chunks(s); + return; + } + sf_collector_append(s->collector, bytes); + scr_bytes_release(bytes); sf_chunk_release(c); } sf_stream_finish_close(s); @@ -837,37 +895,47 @@ static void sf_stream_drain(SfStream *s) { else if (s->closed) sf_collector_finish(s); return; } - if (s->reader && s->reader->pending) { - if (s->head) { + if (s->reader) { + while (s->reader->pending_head && s->head) { SfChunk *c = s->head; s->head = c->next; if (!s->head) s->tail = NULL; - sf_reader_fulfill(s->reader, false, c->bytes); + sf_reader_fulfill_one(s->reader, false, c->value); sf_chunk_release(c); - sf_stream_finish_close(s); - } else if (s->error) { - sf_reader_reject(s->reader, s->error); - } else if (s->closed) { - sf_reader_fulfill(s->reader, true, NULL); } + sf_stream_finish_close(s); + if (s->error) { + sf_reader_reject_all(s->reader, s->error); + } else if (s->closed) { + while (s->reader->pending_head) { + sf_reader_fulfill_one(s->reader, true, NULL); + } + } + return; } sf_stream_finish_close(s); } -static void sf_stream_enqueue_bytes(SfStream *s, ScrBytes *bytes) { +static void sf_stream_enqueue_value(SfStream *s, ScrDyn *value) { if (s->close_requested || s->closed || s->error) { sf_type_error("Invalid state: the ReadableStream is already closed"); return; } SfChunk *c = calloc(1, sizeof *c); if (!c) sf_oom(); - c->bytes = scr_bytes_copy(bytes); + c->value = scr_dyn_retain(value); if (s->tail) s->tail->next = c; else s->head = c; s->tail = c; sf_stream_drain(s); } +static void sf_stream_enqueue_bytes(SfStream *s, ScrBytes *bytes) { + ScrDyn *value = scr_dyn_new_bytes_copy(bytes); + sf_stream_enqueue_value(s, value); + scr_dyn_release(value); +} + static void sf_stream_close(SfStream *s) { if (s->close_requested || s->closed || s->error) return; s->close_requested = true; @@ -878,6 +946,7 @@ static void sf_stream_close(SfStream *s) { static void sf_stream_error(SfStream *s, ScrDyn *reason) { if (s->closed || s->error) return; s->error = scr_dyn_retain(reason); + sf_stream_drop_chunks(s); sf_stream_drain(s); if (s->reader) { sf_reject_promise_reason(s->reader->closed, reason); @@ -909,7 +978,7 @@ static void sf_stream_pull_wait_entry(ScrFiber *self, void *arg) { } else { bool again = s->pull_again || s->request_owner || - (s->reader && s->reader->pending); + (s->reader && s->reader->pending_head); s->pull_again = false; if (again && !s->close_requested && !s->closed && !s->error) { sf_stream_pull(s); @@ -938,7 +1007,7 @@ static void sf_stream_start_wait_entry(ScrFiber *self, void *arg) { s->started = true; sf_stream_drain(s); if (!s->close_requested && !s->closed && !s->error && - (s->request_owner || (s->reader && s->reader->pending))) { + (s->request_owner || (s->reader && s->reader->pending_head))) { sf_stream_pull(s); sf_stream_drain(s); } @@ -990,11 +1059,11 @@ static void sf_stream_pull(SfStream *s) { } scr_dyn_release(r); s->pulling = false; - if (!s->pull_again || s->close_requested || s->closed || s->error) { - s->pull_again = false; - return; - } + bool again = + s->pull_again || s->request_owner || + (s->reader && s->reader->pending_head); s->pull_again = false; + if (!again || s->close_requested || s->closed || s->error) return; } } @@ -1008,6 +1077,9 @@ static SfReader *sf_stream_get_reader(SfStream *s) { r->rc = 1; r->stream = s; r->closed = scr_promise_new(); + /* A reader's internal closed capability never reports as an unhandled + * rejection merely because user code does not read the property. */ + scr_promise_mark_handled(r->closed); if (s->closed) scr_promise_fulfill_void(r->closed); else if (s->error) sf_reject_promise_reason(r->closed, s->error); s->reader = sf_reader_retain(r); @@ -1022,15 +1094,15 @@ static ScrPromise *sf_reader_read(SfReader *r) { scr_promise_reject_pending(p); return p; } - if (r->pending) { - sf_type_error("A read request is already pending"); - scr_promise_reject_pending(p); - return p; - } s->disturbed = true; - r->pending = scr_promise_retain(p); + SfReadRequest *request = calloc(1, sizeof *request); + if (!request) sf_oom(); + request->promise = scr_promise_retain(p); + if (r->pending_tail) r->pending_tail->next = request; + else r->pending_head = request; + r->pending_tail = request; sf_stream_drain(s); - if (r->pending && !s->head && !s->closed && !s->error) { + if (r->pending_head && !s->head && !s->closed && !s->error) { sf_stream_pull(s); sf_stream_drain(s); } @@ -1040,13 +1112,26 @@ static ScrPromise *sf_reader_read(SfReader *r) { static void sf_reader_release_lock(SfReader *r) { SfStream *s = r->stream; if (!s || s->reader != r) return; - if (r->pending) { - sf_type_error("Cannot release a reader with a pending read"); - return; + + ScrPromise *old_closed = r->closed; + if (!s->closed && !s->error) { + sf_type_error("Invalid state: Reader released"); + scr_promise_reject_pending(old_closed); + } else { + r->closed = scr_promise_new(); + scr_promise_mark_handled(r->closed); + sf_type_error("Invalid state: Reader released"); + scr_promise_reject_pending(r->closed); + scr_promise_release(old_closed); } + + sf_reader_reject_message_all(r, "Invalid state: Releasing reader"); + + size_t handle_refs = r->handle_refs; s->reader = NULL; r->stream = NULL; sf_reader_release(r); /* stream's ownership */ + while (handle_refs-- > 0) sf_stream_release(s); } static void sf_stream_cancel_wait_entry(ScrFiber *self, void *arg) { @@ -1199,23 +1284,16 @@ ScrDyn *scr_fetch_stream_new(ScrDyn *source) { } ScrDyn *scr_fetch_stream_from(ScrDyn *iterable) { - if (!iterable || iterable->kind != SCR_DYN_ARR) { - sf_type_error("ReadableStream.from currently requires an array"); - return NULL; - } + ScrDyn *packed = scr_dyn_iter_pack(iterable, NULL); + if (!packed) return NULL; SfStream *s = sf_stream_new_native(); - for (size_t i = 0; i < iterable->v.arr.len; i++) { - ScrBytes *b = sf_chunk_bytes(iterable->v.arr.items[i]); - if (!b) { - sf_stream_release(s); - return NULL; - } - sf_stream_enqueue_bytes(s, b); - scr_bytes_release(b); + for (size_t i = 0; i < packed->v.arr.len; i++) { + sf_stream_enqueue_value(s, packed->v.arr.items[i]); } sf_stream_close(s); ScrDyn *out = scr_dyn_new_handle(s, SCR_DYNH_WEB_STREAM); sf_stream_release(s); + scr_dyn_release(packed); return out; } @@ -1314,10 +1392,7 @@ static ScrDyn *sf_controller_invoke(void *ptr, ScrDyn *self, sf_type_error("ReadableStream controller.enqueue requires one chunk"); return NULL; } - ScrBytes *bytes = sf_chunk_bytes(args[0]); - if (!bytes) return NULL; - sf_stream_enqueue_bytes(s, bytes); - scr_bytes_release(bytes); + sf_stream_enqueue_value(s, args[0]); if (scr_exc_pending()) return NULL; return scr_dyn_retain(scr_dyn_undefined()); } @@ -2157,6 +2232,50 @@ static ScrStr *sf_method(const ScrDyn *init) { return method; } +static bool sf_redirect_mode(const ScrDyn *init, int *out) { + const ScrDyn *value = + init && init->kind == SCR_DYN_OBJ + ? scr_dyn_obj_get(init, "redirect", 8) + : NULL; + if (!value || value->kind == SCR_DYN_UNDEF) { + *out = SF_REDIRECT_FOLLOW; + return true; + } + ScrStr *mode = scr_dyn_string_coerce_js(value); + if (!mode) return false; + if (mode->len == 6 && memcmp(mode->data, "follow", 6) == 0) { + *out = SF_REDIRECT_FOLLOW; + scr_str_release(mode); + return true; + } + if (mode->len == 5 && memcmp(mode->data, "error", 5) == 0) { + *out = SF_REDIRECT_ERROR; + scr_str_release(mode); + return true; + } + if (mode->len == 6 && memcmp(mode->data, "manual", 6) == 0) { + *out = SF_REDIRECT_MANUAL; + scr_str_release(mode); + return true; + } + static const char prefix[] = "undefined: "; + static const char suffix[] = + " is not an accepted type. Expected one of follow, manual, error."; + size_t message_len = + sizeof prefix - 1 + mode->len + sizeof suffix - 1; + char *message = malloc(message_len + 1); + if (!message) sf_oom(); + memcpy(message, prefix, sizeof prefix - 1); + memcpy(message + sizeof prefix - 1, mode->data, mode->len); + memcpy(message + sizeof prefix - 1 + mode->len, + suffix, sizeof suffix - 1); + message[message_len] = '\0'; + scr_str_release(mode); + sf_type_error(message); + free(message); + return false; +} + static ScrPromise *sf_reject_now(ScrPromise *promise, const char *message) { if (!scr_exc_pending()) sf_type_error(message); scr_promise_reject_pending(promise); @@ -2170,7 +2289,19 @@ static void sf_stream_request_flush(SfStream *s) { SfChunk *c = s->head; s->head = c->next; if (!s->head) s->tail = NULL; - scr_http_client_write_bytes(t->client, c->bytes); + ScrBytes *bytes = sf_chunk_bytes(c->value); + if (!bytes) { + sf_chunk_release(c); + sf_stream_drop_chunks(s); + ScrCaught *caught = scr_exc_take(); + ScrDyn *reason = scr_caught_to_dyn(caught); + sf_transfer_stream_error(t, reason); + scr_dyn_release(reason); + scr_caught_release(caught); + return; + } + scr_http_client_write_bytes(t->client, bytes); + scr_bytes_release(bytes); sf_chunk_release(c); } if (s->error) { @@ -2257,7 +2388,7 @@ static void sf_on_res_error(ScrClosure *cb, ScrStr *message) { } static bool sf_response_has_null_body(const SfTransfer *t, int status) { - return sf_eq_ci(t->method, "head") || status == 101 || status == 103 || + return sf_eq_ci(t->method, "head") || status == 101 || status == 204 || status == 205 || status == 304; } @@ -2279,18 +2410,29 @@ static void sf_on_response(ScrClosure *cb, ScrHttpReq *res) { ScrStr *location = scr_http_req_header(res, name); scr_str_release(name); if (location) { - ScrHttpClientReq *old = t->client; - t->client = NULL; - bool follow = sf_redirect(t, status, location); - scr_str_release(location); - if (old) { - scr_http_client_destroy(old); - scr_http_client_release(old); + if (t->redirect_mode == SF_REDIRECT_MANUAL) { + scr_str_release(location); + } else { + ScrHttpClientReq *old = t->client; + t->client = NULL; + if (old) { + scr_http_client_destroy(old); + scr_http_client_release(old); + } + if (t->redirect_mode == SF_REDIRECT_ERROR) { + sf_reject(t, "fetch failed"); + scr_str_release(location); + scr_http_req_release(res); + sf_release(t); + return; + } + bool follow = sf_redirect(t, status, location); + scr_str_release(location); + if (follow) sf_start_hop(t); + scr_http_req_release(res); + sf_release(t); + return; } - if (follow) sf_start_hop(t); - scr_http_req_release(res); - sf_release(t); - return; } } bool null_body = sf_response_has_null_body(t, status); @@ -2543,6 +2685,10 @@ ScrPromise *scr_fetch_static(ScrStr *url, ScrDyn *init) { init->kind != SCR_DYN_UNDEF && init->kind != SCR_DYN_NULL) { return sf_reject_now(promise, "fetch failed"); } + int redirect_mode; + if (!sf_redirect_mode(init, &redirect_mode)) { + return sf_reject_now(promise, "fetch failed"); + } const ScrDyn *signal_dyn = init && init->kind == SCR_DYN_OBJ ? scr_dyn_obj_get(init, "signal", 6) @@ -2560,6 +2706,20 @@ ScrPromise *scr_fetch_static(ScrStr *url, ScrDyn *init) { scr_promise_reject_pending(promise); return promise; } + if (u->userinfo->len > 0) { + static const char prefix[] = + "Request cannot be constructed from a URL that includes credentials: "; + size_t message_len = sizeof prefix - 1 + url->len; + char *message = malloc(message_len + 1); + if (!message) sf_oom(); + memcpy(message, prefix, sizeof prefix - 1); + memcpy(message + sizeof prefix - 1, url->data, url->len); + message[message_len] = '\0'; + sf_type_error(message); + free(message); + scr_url_release(u); + return sf_reject_now(promise, "fetch failed"); + } bool https = sf_eq_ci(u->scheme, "https"); if (!https && !sf_eq_ci(u->scheme, "http")) { scr_url_release(u); @@ -2657,6 +2817,7 @@ ScrPromise *scr_fetch_static(ScrStr *url, ScrDyn *init) { t->method = method; t->headers = headers; t->body = body_present ? scr_dyn_retain((ScrDyn *)body) : NULL; + t->redirect_mode = redirect_mode; if (signal) { t->signal = sf_signal_retain(signal); t->signal_watch = @@ -2686,8 +2847,8 @@ static const ScrDynHandleOps sf_stream_ops = { "ReadableStream", &sf_stream_retain_v, &sf_stream_release_v, &sf_stream_invoke, &sf_stream_get, &sf_no_set, NULL}; static const ScrDynHandleOps sf_reader_ops = { - "ReadableStreamDefaultReader", &sf_reader_retain_v, - &sf_reader_release_v, &sf_reader_invoke, &sf_reader_get, + "ReadableStreamDefaultReader", &sf_reader_handle_retain_v, + &sf_reader_handle_release_v, &sf_reader_invoke, &sf_reader_get, &sf_no_set, NULL}; static const ScrDynHandleOps sf_controller_ops = { "ReadableStreamDefaultController", &sf_stream_retain_v, diff --git a/packages/runtime/src/scr_http.c b/packages/runtime/src/scr_http.c index 21368f62b..c62f671c6 100644 --- a/packages/runtime/src/scr_http.c +++ b/packages/runtime/src/scr_http.c @@ -2608,7 +2608,20 @@ static bool scr_http_client_parse_head(ScrHttpConn *conn, size_t head_len) { return false; } - /* framing: HEAD requests and 204/304 responses have NO body; chunked + /* + * Informational responses do not settle the request. Node emits an + * `information` event for these and keeps parsing until the final + * response; this runtime has no information listener surface yet, so + * consume the head and continue. 101 is the upgrade case below. + */ + if (status >= 100 && status < 200 && status != 101) { + memmove(conn->buf, conn->buf + head_len, conn->len - head_len); + conn->len -= head_len; + scr_http_req_release(res); + return true; + } + + /* framing: HEAD requests and 204/205/304 responses have NO body; chunked * wins over Content-Length; neither means EOF-delimited */ bool chunked = false; long long content_length = -1; @@ -2623,7 +2636,8 @@ static bool scr_http_client_parse_head(ScrHttpConn *conn, size_t head_len) { } } bool head_req = c->method->len == 4 && memcmp(c->method->data, "HEAD", 4) == 0; - bool no_body = head_req || status == 204 || status == 304; + bool no_body = + head_req || status == 204 || status == 205 || status == 304; /* 101 Switching Protocols: 'upgrade' fires INSTEAD of 'response' with * (res, socket, head) — the parser steps aside and the raw socket is diff --git a/tests/fixtures/fetch/servers.mjs b/tests/fixtures/fetch/servers.mjs index f47823f9a..e930661cf 100644 --- a/tests/fixtures/fetch/servers.mjs +++ b/tests/fixtures/fetch/servers.mjs @@ -13,8 +13,9 @@ * form: a plain `GET /__count` to the proxy (a relative-path request no * real proxy client sends) answers the current count. * - * Routes: /text /json /post-echo /header-echo /request-defaults /redirect /slow /drip - * /chunked /gzip /deflate /status-meta /no-content /sse, 404 for the rest; + * Routes: /text /json /post-echo /header-echo /request-defaults /redirect + * /early-hints /invalid-utf8 /slow /drip /chunked /gzip /deflate + * /status-meta /no-content /sse, 404 for the rest; * the proxy relays absolute-URI requests and CONNECT tunnels, counting one * per proxied request either way. */ import { createServer, request } from "node:http"; @@ -66,6 +67,15 @@ export async function startFetchServers() { } else if (url === "/redirect") { res.writeHead(302, { location: "/text" }); res.end(); + } else if (url === "/early-hints") { + res.writeEarlyHints({ + link: "; rel=preload; as=style", + }); + res.writeHead(200, { "content-type": "text/plain" }); + res.end("final"); + } else if (url === "/invalid-utf8") { + res.writeHead(200, { "content-type": "text/plain; charset=utf-8" }); + res.end(Buffer.from([0x61, 0xc3, 0x28, 0x62])); } else if (url === "/slow") { // Answers after 1500ms: the abort/timeout cases cancel long before. setTimeout(() => { diff --git a/tests/fixtures/fetch/static-stream/main.mts b/tests/fixtures/fetch/static-stream/main.mts index f8bca9852..2f02169d9 100644 --- a/tests/fixtures/fetch/static-stream/main.mts +++ b/tests/fixtures/fetch/static-stream/main.mts @@ -58,6 +58,78 @@ const promisedPost = await fetch(`${process.argv[2]}/post-echo`, { }); console.log(await promisedPost.json()); +const temporaryRead = await new ReadableStream({ + start(controller) { + controller.enqueue(Buffer.from("temporary")); + controller.close(); + }, +}).getReader().read(); +console.log( + "temporary reader:", + temporaryRead.done ? "done" : new TextDecoder().decode(temporaryRead.value), +); + +let concurrentValue = 0; +const concurrentReader = new ReadableStream({ + async pull(controller) { + await new Promise((resolve) => setTimeout(resolve, 5)); + concurrentValue++; + controller.enqueue(concurrentValue); + if (concurrentValue === 2) controller.close(); + }, +}).getReader(); +async function readConcurrent(): Promise> { + return await concurrentReader.read(); +} +const concurrentFirstPromise = readConcurrent(); +const concurrentSecondPromise = readConcurrent(); +const concurrentFirst = await concurrentFirstPromise; +const concurrentSecond = await concurrentSecondPromise; +console.log( + "concurrent reads:", + concurrentFirst.value, + concurrentSecond.value, +); + +const releasedReader = new ReadableStream({ + async start() { + await new Promise((resolve) => setTimeout(resolve, 5)); + }, +}).getReader(); +const releasedRead: unknown = releasedReader.read(); +const oldReleasedClosed: unknown = releasedReader.closed; +releasedReader.releaseLock(); +const newReleasedClosed: unknown = releasedReader.closed; +try { + await oldReleasedClosed; +} catch (error) { + const caught = error as Error; + console.log("released old closed:", caught.name, caught.message); +} +try { + await releasedRead; +} catch (error) { + const caught = error as Error; + console.log("released read:", caught.name, caught.message); +} +try { + await newReleasedClosed; +} catch (error) { + const caught = error as Error; + console.log("released new closed:", caught.name, caught.message); +} + +const stringReader = ReadableStream.from("😀a").getReader(); +const stringFirst = await stringReader.read(); +const stringSecond = await stringReader.read(); +const stringDone = await stringReader.read(); +console.log( + "stream from string:", + stringFirst.value, + stringSecond.value, + stringDone.done, +); + const streamed = await fetch(`${process.argv[2]}/chunked`); const reader = streamed.body!.getReader(); const chunks: Uint8Array[] = []; diff --git a/tests/fixtures/fetch/static/main.mts b/tests/fixtures/fetch/static/main.mts index b58e30a93..55e75807f 100644 --- a/tests/fixtures/fetch/static/main.mts +++ b/tests/fixtures/fetch/static/main.mts @@ -61,6 +61,43 @@ console.log( await redirected.text(), ); +const manualRedirect = await fetch(`${process.argv[2]}/redirect`, { + redirect: "manual", +}); +console.log( + "manual redirect:", + manualRedirect.status, + manualRedirect.redirected, + manualRedirect.url.endsWith("/redirect"), + manualRedirect.headers.get("location"), + JSON.stringify(await manualRedirect.text()), +); + +try { + await fetch(`${process.argv[2]}/redirect`, { redirect: "error" }); +} catch (error) { + const caught = error as Error; + console.log("error redirect:", caught.name, caught.message); +} + +try { + const credentialUrl = + `http://user:pass@${process.argv[2].slice("http://".length)}/text`; + await fetch(credentialUrl); +} catch (error) { + const caught = error as Error; + console.log("credential URL:", caught.name, caught.message); +} + +console.log( + "early hints:", + await (await fetch(`${process.argv[2]}/early-hints`)).text(), +); +console.log( + "invalid utf8:", + JSON.stringify(await (await fetch(`${process.argv[2]}/invalid-utf8`)).text()), +); + const statusMeta = await fetch(`${process.argv[2]}/status-meta`); console.log("status text:", statusMeta.status, statusMeta.statusText); diff --git a/tests/fixtures/node-types/fetch-static.ts b/tests/fixtures/node-types/fetch-static.ts index a4dfca92e..afdfaa1a7 100644 --- a/tests/fixtures/node-types/fetch-static.ts +++ b/tests/fixtures/node-types/fetch-static.ts @@ -11,6 +11,7 @@ const init: RequestInit = { method: "POST", body, duplex: "half", + redirect: "manual", signal: AbortSignal.timeout(100), };