Clone Response and Request bodies by teeing their streams

Response.clone() and Request.clone() on checked-dynamic handles answered
'res.clone is not a function' (hono's etag middleware hashes
res.clone().body). The runtime now implements the streams spec's
ReadableStreamDefaultTee over native streams and clones the body with it,
so the original and the clone each read every chunk. A used or locked body
throws undici's TypeErrors, the clone's headers are a copy with the same
guard, and a cloned Request's signal follows the original's.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7742-response-clone-tee.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7742-response-clone-tee.ts"
+ ],
+ "diags": []
+}

packages/runtime/src/scr_fetch.c

@@ -1855,6 +1855,210 @@ static ScrPromise *sf_stream_collect(SfStream *s, int mode) {
   return p;
 }
 
+/* ── ReadableStreamDefaultTee ────────────────────────────────────── */
+
+/*
+ * The streams spec's ReadableStreamDefaultTee over a native stream: the
+ * tee locks the source with a reader and feeds two native branches whose
+ * pull and cancel algorithms are native functions. A pull on either branch
+ * reads one chunk from the source and enqueues it into every branch that
+ * was not canceled; canceling both branches cancels the source with both
+ * reasons. Response.clone and Request.clone use it (undici's cloneBody).
+ *
+ * Ownership: the tee owns the source, its reader and both branches; each
+ * branch's callbacks and every pending read own the tee. Closing, erroring
+ * or canceling a branch drops its callbacks (sf_stream_drop_source_callbacks),
+ * and fetch teardown drops the callbacks of branches nobody finished.
+ */
+typedef struct SfTee {
+  size_t rc;
+  SfStream *source;
+  SfReader *reader;
+  SfStream *branch[2];
+  bool reading;
+  bool read_again;
+  bool canceled[2];
+  ScrDyn *reason[2];
+  ScrPromise *cancel_promise;
+} SfTee;
+
+static SfTee *sf_tee_retain(SfTee *t) {
+  t->rc++;
+  return t;
+}
+
+static void sf_tee_release(SfTee *t) {
+  if (!t || --t->rc > 0) return;
+  sf_stream_release(t->branch[0]);
+  sf_stream_release(t->branch[1]);
+  sf_reader_release(t->reader);
+  sf_stream_release(t->source);
+  scr_dyn_release(t->reason[0]);
+  scr_dyn_release(t->reason[1]);
+  scr_promise_release(t->cancel_promise);
+  free(t);
+}
+
+static void *sf_tee_retain_v(void *p) { return sf_tee_retain((SfTee *)p); }
+static void sf_tee_release_v(void *p) { sf_tee_release((SfTee *)p); }
+
+static ScrBox *sf_tee_box(SfTee *t) {
+  ScrBox *box = scr_box_new_obj(&sf_tee_retain_v, &sf_tee_release_v, NULL);
+  scr_box_set_ref(box, sf_tee_retain(t));
+  return box;
+}
+
+static void sf_tee_pull(SfTee *t);
+
+/* The read request's chunk steps run in a microtask (spec: "queue a
+ * microtask"), after the read that produced the chunk settles. */
+static void sf_tee_chunk_task(ScrClosure *cb) {
+  SfTee *t = (SfTee *)scr_box_get_ref(cb->caps[0]);
+  ScrDyn *chunk = (ScrDyn *)scr_box_get_ref(cb->caps[1]);
+  t->read_again = false;
+  for (int i = 0; i < 2; i++) {
+    if (t->canceled[i]) continue;
+    SfStream *b = t->branch[i];
+    if (b->close_requested || b->closed || b->error) continue;
+    sf_stream_enqueue_value(b, chunk);
+    if (scr_exc_pending()) scr_exc_clear();
+  }
+  t->reading = false;
+  if (t->read_again) sf_tee_pull(t);
+  scr_dyn_release(chunk);
+  sf_tee_release(t);
+}
+
+static void sf_tee_sink_chunk(void *ctx, ScrDyn *value) {
+  SfTee *t = ctx;
+  ScrClosure *cb = scr_closure_new((void *)&sf_tee_chunk_task, 2);
+  cb->caps[0] = sf_tee_box(t);
+  cb->caps[1] = scr_box_new_obj(&scr_dyn_retain_v, &scr_dyn_release_v, NULL);
+  scr_box_set_ref(cb->caps[1], scr_dyn_retain(value));
+  scr_queue_microtask(cb);
+  sf_tee_release(t); /* the read request's reference */
+}
+
+static void sf_tee_sink_close(void *ctx) {
+  SfTee *t = ctx;
+  t->reading = false;
+  for (int i = 0; i < 2; i++) {
+    if (!t->canceled[i]) sf_stream_close(t->branch[i]);
+  }
+  if (!t->canceled[0] || !t->canceled[1]) {
+    scr_promise_fulfill_void(t->cancel_promise);
+  }
+  sf_tee_release(t);
+}
+
+static void sf_tee_sink_error(void *ctx, ScrDyn *reason) {
+  SfTee *t = ctx;
+  t->reading = false;
+  if (reason) {
+    /* The spec errors both branches when the source reader's closed
+     * promise rejects; a native read request learns it here. */
+    sf_stream_error(t->branch[0], reason);
+    sf_stream_error(t->branch[1], reason);
+    if (!t->canceled[0] || !t->canceled[1]) {
+      scr_promise_fulfill_void(t->cancel_promise);
+    }
+  }
+  sf_tee_release(t);
+}
+
+static const ScrFetchReadSink sf_tee_sink = {
+    &sf_tee_sink_chunk, &sf_tee_sink_close, &sf_tee_sink_error};
+
+static void sf_tee_pull(SfTee *t) {
+  if (t->reading) {
+    t->read_again = true;
+    return;
+  }
+  t->reading = true;
+  sf_tee_retain(t); /* owned by the read request */
+  if (!sf_reader_read_request(t->reader, NULL, &sf_tee_sink, t)) {
+    scr_exc_clear();
+    t->reading = false;
+    sf_tee_release(t);
+  }
+}
+
+static ScrDyn *sf_tee_pull_thunk(ScrClosure *cb, ScrDyn *const *args,
+                                 size_t argc) {
+  (void)args;
+  (void)argc;
+  SfTee *t = (SfTee *)scr_box_get_ref(cb->caps[0]);
+  sf_tee_pull(t);
+  sf_tee_release(t);
+  return scr_dyn_retain(scr_dyn_undefined());
+}
+
+static ScrDyn *sf_tee_cancel(SfTee *t, int i, ScrDyn *reason) {
+  t->canceled[i] = true;
+  scr_dyn_release(t->reason[i]);
+  t->reason[i] = scr_dyn_retain(reason);
+  if (t->canceled[0] && t->canceled[1]) {
+    ScrDyn *composite = scr_dyn_new_arr();
+    scr_dyn_arr_push(composite, scr_dyn_retain(t->reason[0]));
+    scr_dyn_arr_push(composite, scr_dyn_retain(t->reason[1]));
+    ScrPromise *result = sf_stream_cancel(t->source, composite, true);
+    scr_dyn_release(composite);
+    ScrDyn *boxed = scr_dyn_new_promise(result);
+    scr_promise_release(result);
+    sf_then_adopt(t->cancel_promise, boxed);
+    scr_dyn_release(boxed);
+  }
+  return scr_dyn_new_promise(t->cancel_promise);
+}
+
+static ScrDyn *sf_tee_cancel_thunk_0(ScrClosure *cb, ScrDyn *const *args,
+                                     size_t argc) {
+  SfTee *t = (SfTee *)scr_box_get_ref(cb->caps[0]);
+  ScrDyn *out = sf_tee_cancel(t, 0, argc ? args[0] : scr_dyn_undefined());
+  sf_tee_release(t);
+  return out;
+}
+
+static ScrDyn *sf_tee_cancel_thunk_1(ScrClosure *cb, ScrDyn *const *args,
+                                     size_t argc) {
+  SfTee *t = (SfTee *)scr_box_get_ref(cb->caps[0]);
+  ScrDyn *out = sf_tee_cancel(t, 1, argc ? args[0] : scr_dyn_undefined());
+  sf_tee_release(t);
+  return out;
+}
+
+static ScrDyn *sf_tee_func(SfTee *t, ScrDynThunk thunk, const char *name) {
+  ScrClosure *cb = scr_closure_new((void *)thunk, 1);
+  cb->caps[0] = sf_tee_box(t);
+  return scr_dyn_new_func(cb, thunk, 1, "(controller)", name);
+}
+
+/* Tees an unlocked, undisturbed stream: +1 branches in out[0] and out[1].
+ * The source stays locked by the tee. */
+static void sf_stream_tee(SfStream *s, SfStream *out[2]) {
+  SfTee *t = calloc(1, sizeof *t);
+  if (!t) sf_oom();
+  t->rc = 1;
+  t->source = sf_stream_retain(s);
+  t->reader = sf_stream_get_reader(s);
+  t->cancel_promise = scr_promise_new();
+  scr_promise_mark_handled(t->cancel_promise);
+  for (int i = 0; i < 2; i++) {
+    SfStream *b = sf_stream_new_native();
+    b->pull_cb = sf_tee_func(t, &sf_tee_pull_thunk, "pull");
+    b->cancel_cb = sf_tee_func(
+        t, i == 0 ? &sf_tee_cancel_thunk_0 : &sf_tee_cancel_thunk_1, "cancel");
+    b->source_this = scr_dyn_retain(scr_dyn_undefined());
+    sf_stream_track_callbacks(b);
+    t->branch[i] = sf_stream_retain(b);
+    out[i] = b;
+  }
+  /* The branches start at once; a branch with room then pulls. */
+  sf_stream_schedule_initial_pull(out[0]);
+  sf_stream_schedule_initial_pull(out[1]);
+  sf_tee_release(t);
+}
+
 ScrDyn *scr_fetch_stream_new(ScrDyn *source) {
   ScrDyn *source_view = source;
   ScrDyn *materialized = NULL;
@@ -2729,12 +2933,54 @@ static ScrPromise *sf_response_collect(SfResponse *r, int mode) {
   return p;
 }
 
+/* The body and metadata copy of Response.clone and Request.clone (undici
+ * cloneResponse/cloneBody): the body tees, so the original and the clone
+ * each keep one branch, and the clone's headers are a copy with the same
+ * guard. NULL with a TypeError pending when the body is unusable. */
+static SfResponse *sf_response_clone(SfResponse *r, const char *unusable) {
+  if (!r->null_body &&
+      (r->body->disturbed || r->body->reader || r->body->internal_lock)) {
+    sf_type_error(unusable);
+    return NULL;
+  }
+  SfResponse *c = calloc(1, sizeof *c);
+  if (!c) sf_oom();
+  c->rc = 1;
+  if (r->null_body) {
+    c->body = sf_stream_new_native();
+    sf_stream_close(c->body);
+  } else {
+    SfStream *branches[2];
+    sf_stream_tee(r->body, branches);
+    sf_stream_release(r->body);
+    r->body = branches[0];
+    c->body = branches[1];
+  }
+  c->headers = sf_headers_new(scr_arr_slice(r->headers->pairs, 0, INFINITY),
+                              r->headers->immutable);
+  c->url = scr_str_retain(r->url);
+  c->status_text = scr_str_retain(r->status_text);
+  c->status = r->status;
+  c->redirected = r->redirected;
+  c->null_body = r->null_body;
+  c->light_length = r->light_length;
+  return c;
+}
+
 static ScrDyn *sf_response_invoke(void *ptr, ScrDyn *self,
                                   const char *method,
                                   ScrDyn *const *args, size_t argc,
                                   const char *what) {
   SfResponse *r = ptr;
   sf_response_materialize(r);
+  if (strcmp(method, "clone") == 0) {
+    SfResponse *c = sf_response_clone(
+        r, "Response.clone: Body has already been consumed.");
+    if (!c) return NULL;
+    ScrDyn *out = scr_dyn_new_handle(c, SCR_DYNH_FETCH_RESPONSE);
+    sf_response_release(c);
+    return out;
+  }
   int mode = -1;
   if (strcmp(method, "json") == 0) mode = SF_COLLECT_JSON;
   else if (strcmp(method, "text") == 0) mode = SF_COLLECT_TEXT;
@@ -5153,8 +5399,37 @@ static ScrDyn *sf_request_get(void *ptr, const char *key, size_t len) {
   if (sf_name(key, len, "body") || sf_name(key, len, "bodyUsed") || sf_name(key, len, "headers")) return sf_response_get(r->payload, key, len);
   return NULL;
 }
+/* Request.clone: the payload clones as a Response does, and the clone's
+ * signal follows the original's (undici: an AbortController aborted with
+ * the original signal's reason, at once when it already aborted). */
+static ScrDyn *sf_request_clone(SfRequest *r) {
+  SfResponse *payload = sf_response_clone(r->payload, "unusable");
+  if (!payload) return NULL;
+  ScrDyn *sources = scr_dyn_new_arr();
+  scr_dyn_arr_push(sources,
+                   scr_dyn_new_handle(r->signal, SCR_DYNH_ABORT_SIGNAL));
+  ScrDyn *follower = scr_fetch_abort_any(sources);
+  scr_dyn_release(sources);
+  if (!follower) {
+    sf_response_release(payload);
+    return NULL;
+  }
+  SfRequest *c = calloc(1, sizeof *c);
+  if (!c) sf_oom();
+  c->rc = 1;
+  c->payload = payload;
+  c->url = scr_str_retain(r->url);
+  c->method = scr_str_retain(r->method);
+  c->redirect = scr_str_retain(r->redirect);
+  c->signal = sf_signal_retain((SfSignal *)follower->v.handle.ptr);
+  scr_dyn_release(follower);
+  ScrDyn *out = scr_dyn_new_handle(c, SCR_DYNH_FETCH_REQUEST);
+  sf_request_release_v(c);
+  return out;
+}
 static ScrDyn *sf_request_invoke(void *ptr, ScrDyn *self, const char *method, ScrDyn *const *args, size_t argc, const char *what) {
   SfRequest *r = ptr;
+  if (strcmp(method, "clone") == 0) return sf_request_clone(r);
   return sf_response_invoke(r->payload, self, method, args, argc, what);
 }
 
@@ -5668,7 +5943,7 @@ static const ScrDynHandleMethod sf_reader_methods[] = {
     {NULL, 0, false}};
 static const ScrDynHandleMethod sf_body_methods[] = {
     {"json", 0, true}, {"text", 0, true}, {"bytes", 0, true},
-    {"arrayBuffer", 0, true}, {NULL, 0, false}};
+    {"arrayBuffer", 0, true}, {"clone", 0, false}, {NULL, 0, false}};
 static const ScrDynHandleMethod sf_headers_methods[] = {
     {"append", 2, false}, {"delete", 1, false}, {"entries", 0, false},
     {"forEach", 1, false}, {"get", 1, false}, {"getSetCookie", 0, false},

tests/corpus/7742-response-clone-tee.ts

@@ -0,0 +1,154 @@
+// Response.clone and Request.clone on checked-dynamic handles: the body
+// tees, so the original and the clone each read every chunk; a clone of a
+// used or locked body throws; canceling both branches cancels the source
+// with both reasons; the clone's headers are a copy.
+const enc = new TextEncoder();
+const dec = new TextDecoder();
+
+function describe(e: unknown): string {
+  const err = e as { name?: string; message?: string };
+  return `${String(err.name)}: ${String(err.message)}`;
+}
+
+async function readAll(stream: any): Promise<string> {
+  const reader = stream.getReader();
+  let out = "";
+  for (;;) {
+    const { done, value } = await reader.read();
+    if (done) return out;
+    out += dec.decode(value);
+  }
+}
+
+async function main(): Promise<void> {
+  const r: any = new Response("hello world", { status: 201, headers: { "x-a": "1" } });
+  const c: any = r.clone();
+  console.log(c.status, c.headers.get("x-a"), c.bodyUsed, r.bodyUsed, c.body === r.body);
+  console.log(await c.text(), "|", await r.text());
+  console.log("used", c.bodyUsed, r.bodyUsed);
+  try {
+    r.clone();
+  } catch (e) {
+    console.log("clone used:", describe(e), e instanceof TypeError);
+  }
+
+  const locked: any = new Response("abc");
+  locked.body.getReader();
+  try {
+    locked.clone();
+  } catch (e) {
+    console.log("clone locked:", describe(e));
+  }
+
+  const empty: any = new Response(null, { status: 204, statusText: "Nothing" });
+  const emptyClone: any = empty.clone();
+  console.log("null body", emptyClone.body, emptyClone.status, emptyClone.statusText, JSON.stringify(await emptyClone.text()));
+
+  // Headers are copied: changing the clone's leaves the original's alone.
+  const h: any = new Response("x", { headers: { "x-b": "2" } });
+  const hc: any = h.clone();
+  hc.headers.set("x-b", "3");
+  console.log("headers", h.headers.get("x-b"), hc.headers.get("x-b"), h.headers === hc.headers);
+
+  // A pulled source: both branches see each chunk (the same object).
+  let n = 0;
+  const source: any = new ReadableStream({
+    start(): void {
+      console.log("start");
+    },
+    pull(controller: any): void {
+      n++;
+      console.log("pull", n);
+      if (n <= 3) controller.enqueue(enc.encode(`chunk${n};`));
+      else controller.close();
+    },
+  });
+  const streamed: any = new Response(source);
+  const streamedClone: any = streamed.clone();
+  console.log("cloned");
+  const r1 = streamed.body.getReader();
+  const r2 = streamedClone.body.getReader();
+  const a = await r1.read();
+  const b = await r2.read();
+  console.log("same chunk", a.value === b.value, dec.decode(a.value));
+  console.log("rest 1", await readAllFrom(r1));
+  console.log("rest 2", await readAllFrom(r2));
+
+  // One branch read to the end first; the other buffers.
+  const big: any = new Response(new ReadableStream({
+    start(controller: any): void {
+      controller.enqueue(enc.encode("a"));
+      controller.enqueue(enc.encode("b"));
+      controller.close();
+    },
+  }));
+  const bigClone: any = big.clone();
+  console.log("clone first", await bigClone.text());
+  console.log("then original", await readAll(big.body));
+
+  // Canceling both branches cancels the source with both reasons.
+  const cancelSource: any = new ReadableStream({
+    pull(controller: any): void {
+      controller.enqueue(enc.encode("z"));
+    },
+    cancel(reason: unknown): void {
+      console.log("source cancel", JSON.stringify(reason));
+    },
+  });
+  const cr: any = new Response(cancelSource);
+  const crc: any = cr.clone();
+  const p1 = cr.body.cancel("first");
+  console.log("one canceled");
+  const p2 = crc.body.cancel("second");
+  await p1;
+  await p2;
+  console.log("both canceled", cr.bodyUsed, crc.bodyUsed);
+
+  // An erroring source errors both branches.
+  const failing: any = new ReadableStream({
+    pull(controller: any): void {
+      controller.error(new Error("source failed"));
+    },
+  });
+  const fr: any = new Response(failing);
+  const frc: any = fr.clone();
+  for (const branch of [fr, frc]) {
+    try {
+      await branch.text();
+    } catch (e) {
+      console.log("branch error:", describe(e));
+    }
+  }
+
+  // Request.clone: the body tees and the signal follows the original's.
+  const q: any = new Request("http://example.test/path", { method: "POST", body: "data" });
+  const qc: any = q.clone();
+  console.log("request", qc.method, qc.url, await qc.text(), await q.text(), qc.signal === q.signal);
+  try {
+    q.clone();
+  } catch (e) {
+    console.log("request clone used:", describe(e));
+  }
+  const ac = new AbortController();
+  const q2: any = new Request("http://example.test/", { signal: ac.signal });
+  const q2c: any = q2.clone();
+  console.log("before abort", q2c.signal.aborted);
+  ac.abort("why");
+  console.log("after abort", q2c.signal.aborted, q2c.signal.reason);
+  const q3: any = new Request("http://example.test/", { signal: AbortSignal.abort("early") });
+  console.log("aborted clone", q3.clone().signal.aborted, q3.clone().signal.reason);
+
+  // clone is a method value like the body readers.
+  console.log(typeof r.clone, "clone" in r, r.clone === c.clone);
+}
+
+async function readAllFrom(reader: any): Promise<string> {
+  let out = "";
+  for (;;) {
+    const { done, value } = await reader.read();
+    if (done) return out;
+    out += dec.decode(value);
+  }
+}
+
+await main();

tests/harness/corpus-costs/7742-response-clone-tee.ts.json

@@ -0,0 +1 @@
+{ "plain": 3000, "san": 4000 }