The reactor owns the HTTP/1 frame, so a cancelled write cannot outlive it (#179)

src/core/http_connection.c

@@ -1413,6 +1413,78 @@ bool http_connection_send_raw(http_connection_t *conn,
 }
 /* }}} */
 
+#if defined(ZEND_ASYNC_API_VERSION_NUMBER) && ZEND_ASYNC_API_VERSION_NUMBER >= 0x001900
+/* {{{ http_connection_send_strv_awaited
+ *
+ * Vectored plaintext send the caller waits for. Slots go out in array order;
+ * every reference is consumed whatever happens, including a cancellation while
+ * parked — the reactor holds them until libuv is done, which is exactly what
+ * the awaited single-buffer write cannot promise. */
+bool http_connection_send_strv_awaited(http_connection_t *conn,
+                                       zend_string *const *bufs,
+                                       const unsigned nbufs)
+{
+    /* The one refusal that neither throws nor consumes: unrecognisable after. */
+    ZEND_ASSERT(nbufs > 0);
+
+    if (UNEXPECTED(conn->write_timed_out)) {
+        for (unsigned i = 0; i < nbufs; i++) {
+            zend_string_release(bufs[i]);
+        }
+
+        return false;
+    }
+
+    const uint32_t write_timeout_ms = conn->write_timeout_ms;
+
+    if (write_timeout_ms > 0 && !http_write_timer_arm(conn, write_timeout_ms)) {
+        for (unsigned i = 0; i < nbufs; i++) {
+            zend_string_release(bufs[i]);
+        }
+
+        return false;
+    }
+
+    size_t total = 0;
+
+    for (unsigned i = 0; i < nbufs; i++) {
+        total += ZSTR_LEN(bufs[i]);
+    }
+
+    bool ok_total = false;
+    zend_async_io_req_t *req = ZEND_ASYNC_IO_WRITEV_AWAITED(conn->io, bufs, nbufs);
+
+    if (req != NULL) {
+        const bool ok = async_io_req_await(req, conn->io, write_timeout_ms,
+                                           HTTP_IO_REQ_WRITE, conn->log_state);
+        const bool had_exc = (req->exception != NULL);
+
+        if (had_exc) {
+            OBJ_RELEASE(req->exception);
+            req->exception = NULL;
+        }
+
+        const ssize_t transferred = req->transferred;
+        req->dispose(req);
+        ok_total = ok && !had_exc && transferred == (ssize_t)total;
+    } else {
+        /* Past the guard every refusal has released the slots already. */
+        http_absorb_io_submission_exception(conn, __func__);
+    }
+
+    if (write_timeout_ms > 0) {
+        http_write_timer_stop(conn);
+    }
+
+    if (UNEXPECTED(conn->write_timed_out)) {
+        return false;
+    }
+
+    return ok_total;
+}
+/* }}} */
+#endif /* async API >= 0.25 */
+
 /* {{{ http_connection_send_str_owned
  *
  * Fire-and-forget plaintext send: transfer ownership of @p body to the
@@ -2781,7 +2853,15 @@ void http_handler_coroutine_dispose(zend_coroutine_t *coroutine)
     conn->state = CONN_STATE_SENDING;
 
     if (http_response_is_streaming(Z_OBJ(ctx->response_zv))) {
-        if (!http_response_is_closed(Z_OBJ(ctx->response_zv))) {
+        if (UNEXPECTED(ctx->stream_dead)) {
+            /* A cancellation can cut a frame in half, and the peer has been
+             * promised the bytes its size line named. Sealing that with the
+             * terminator would say the body ended cleanly and hand the
+             * connection on, and the peer would read the terminator as the
+             * first bytes of what it is still waiting for. mark_ended refuses
+             * the same way; this is the path that skips mark_ended. */
+            conn->keep_alive = 0;
+        } else if (!http_response_is_closed(Z_OBJ(ctx->response_zv))) {
             /* Handler fell through without end() — emit the terminator. */
             (void)http_connection_send(conn, "0\r\n\r\n", 5);
         }

src/core/http_connection.h

@@ -461,6 +461,15 @@ bool http_connection_send_batched_writev(http_connection_t *conn,
 bool http_connection_send_strv_owned(http_connection_t *conn,
                                      zend_string * const *bufs, unsigned nbufs);
 
+#if defined(ZEND_ASYNC_API_VERSION_NUMBER) && ZEND_ASYNC_API_VERSION_NUMBER >= 0x001900
+/* Vectored variant the caller waits for. Each slot is an OWNED reference and is
+ * consumed in every outcome, a cancellation while parked included — the reactor
+ * keeps them until libuv is done, so a queued write never points at freed bytes.
+ * Plaintext only — same TLS caveat as send_str_owned. Requires nbufs > 0. */
+bool http_connection_send_strv_awaited(http_connection_t *conn,
+                                       zend_string * const *bufs, unsigned nbufs);
+#endif
+
 /* Outbound backpressure (transport-level, plaintext batched path).
  * pending_bytes = coalesced tail waiting behind the single in-flight
  * batched write — the part that grows under a slow consumer. The

src/http1/http1_stream.c

@@ -59,6 +59,28 @@
 ZEND_STATIC_ASSERT(H1_CHUNK_COALESCE_MAX <= HTTP_TLS_PLAINTEXT_RING_BYTES,
                    "a coalesced frame must fit one TLS plaintext ring cycle");
 
+/* Sends a header block the caller owns, consuming its reference. Where the
+ * reactor can take it over it does, so a cancellation cannot leave a queued
+ * write pointing at a released string; elsewhere the copy is unavoidable. */
+static bool h1_send_headers_owned(http_connection_t *conn, zend_string *headers)
+{
+#if defined(ZEND_ASYNC_API_VERSION_NUMBER) && ZEND_ASYNC_API_VERSION_NUMBER >= 0x001900
+#ifdef HAVE_OPENSSL
+    const bool plaintext = conn->tls == NULL;
+#else
+    const bool plaintext = true;
+#endif
+
+    if (plaintext) {
+        return http_connection_send_strv_awaited(conn, &headers, 1);
+    }
+#endif
+
+    const bool ok = http_connection_send(conn, ZSTR_VAL(headers), ZSTR_LEN(headers));
+    zend_string_release(headers);
+    return ok;
+}
+
 /* The status line and headers of a streaming response, as bytes. Returns NULL
  * when the response is gone or formats to nothing; the caller owns the string.
  * Separate from the send so the first frame can carry the block with it. */
@@ -113,10 +135,7 @@ static bool h1_emit_headers_once(http1_request_ctx_t *ctx)
         return false;
     }
 
-    const bool ok = http_connection_send(ctx->conn, ZSTR_VAL(headers),
-                                         ZSTR_LEN(headers));
-    zend_string_release(headers);
-    return ok;
+    return h1_send_headers_owned(ctx->conn, headers);
 }
 
 /* `nonblocking` is accepted and ignored: HTTP/1 keeps no queue of its own, so
@@ -179,9 +198,7 @@ static int h1_stream_append_chunk(void *opaque, zend_string *chunk,
         zend_string_release(chunk);
 
         if (headers != NULL) {
-            const bool sent = http_connection_send(conn, ZSTR_VAL(headers),
-                                                   ZSTR_LEN(headers));
-            zend_string_release(headers);
+            const bool sent = h1_send_headers_owned(conn, headers);
 
             if (!sent) {
                 ctx->stream_dead = true;
@@ -221,7 +238,51 @@ static int h1_stream_append_chunk(void *opaque, zend_string *chunk,
      * over — today's ABI offers one or the other, never both. */
     const size_t head_len  = headers != NULL ? ZSTR_LEN(headers) : 0;
     const size_t frame_len = head_len + (size_t)header_len + chunk_len + 2;
-    const bool   coalesce  = frame_len <= H1_CHUNK_COALESCE_MAX;
+
+#if defined(ZEND_ASYNC_API_VERSION_NUMBER) && ZEND_ASYNC_API_VERSION_NUMBER >= 0x001900
+    /* Plaintext: the pieces go over as slots the reactor owns — one submit, no
+     * copy of the body, and nothing a cancelled frame could leave dangling.
+     * TLS keeps the copy below: tls_push copies into the BIO ring anyway, and a
+     * vectored write at the socket would put plaintext on a TLS connection. */
+#ifdef HAVE_OPENSSL
+    const bool plaintext = conn->tls == NULL;
+#else
+    const bool plaintext = true;
+#endif
+
+    if (plaintext) {
+        zend_string *slots[4];
+        unsigned     n = 0;
+
+        if (headers != NULL) {
+            slots[n++] = headers;            /* ownership moves to the reactor */
+            headers    = NULL;
+        }
+
+        slots[n++] = zend_string_init(header, (size_t)header_len, 0);
+        slots[n++] = chunk;                  /* the caller's ref, handed over */
+        /* Two bytes per frame rather than one shared literal: the send releases
+         * every slot, and interning at runtime would write the process-wide
+         * permanent table from a worker thread. */
+        slots[n++] = zend_string_init("\r\n", 2, 0);
+
+        const bool sent = http_connection_send_strv_awaited(conn, slots, n);
+
+        if (UNEXPECTED(!sent || EG(exception) != NULL)) {
+            ctx->stream_dead = true;
+            return HTTP_STREAM_APPEND_STREAM_DEAD;
+        }
+
+        if (head_len != 0) {
+            h1_stream_headers_committed(ctx);
+        }
+
+        http_server_on_stream_send(conn->counters, chunk_len);
+        return HTTP_STREAM_APPEND_OK;
+    }
+#endif
+
+    const bool coalesce = frame_len <= H1_CHUNK_COALESCE_MAX;
     bool frame_ok = true;
 
     /* Too large to carry the block along: the headers go out on their own, and

tests/phpt/server/h1/030-h1-cancel-while-parked-in-write.phpt

@@ -0,0 +1,131 @@
+--TEST--
+HttpResponse::write() — a handler cancelled while parked inside a write unwinds cleanly
+--EXTENSIONS--
+true_async_server
+true_async
+sockets
+--FILE--
+<?php
+/* The shape nothing in this suite covered, and the reason a lifetime defect on
+ * this path stayed green for months: a handler parked inside a write, then
+ * cancelled while the write is still in libuv's queue.
+ *
+ * Getting there needs three things at once. The peer must exist and must not
+ * read, so the write parks instead of failing — SO_RCVBUF is shrunk to 4 KiB so
+ * the server's socket buffer fills within one chunk. The cancellation must
+ * arrive while the handler is parked, which means skipping the grace window:
+ * stop() waits shutdown_timeout_s for handlers to finish on their own and only
+ * then cancels the scope, so at the default of five seconds the handler always
+ * wins the race. And the response must be streaming, because that is the only
+ * path where the handler owns the write.
+ *
+ * Three assertions, and each one pins something different. The frames read
+ * before the cancel are checked byte for byte, which is what catches a frame
+ * assembled in the wrong order or with a piece already released. The tail read
+ * after it must carry no terminator: sealing a frame the cancellation cut in
+ * half is what desynchronises the next request on a kept-alive connection. And
+ * the cancellation itself must arrive as HttpException 499 with isWritable()
+ * false afterwards. A write timeout of one second turns a lost wake into a
+ * bounded failure rather than a suite that hangs. */
+
+use TrueAsync\HttpServer;
+use TrueAsync\HttpServerConfig;
+use function Async\spawn;
+use function Async\await;
+use function Async\delay;
+
+require_once __DIR__ . '/../_free_port.inc';
+
+$port = tas_free_port();
+$server = new HttpServer((new HttpServerConfig())
+    ->addListener('127.0.0.1', $port)
+    ->setReadTimeout(30)->setWriteTimeout(1)
+    ->setShutdownTimeout(0));          /* no grace window: cancel at once */
+
+$seen = [];
+
+$server->addHttpHandler(function ($req, $res) use (&$seen) {
+    $res->setStatusCode(200)->setHeader('Content-Type', 'application/octet-stream');
+
+    try {
+        /* Far more than any socket buffer holds; parks after the first chunks. */
+        for ($i = 0; $i < 20000; $i++) {
+            $res->write(str_repeat('x', 2000));
+        }
+        $res->end();
+        $seen['outcome'] = 'finished';
+    } catch (\Throwable $e) {
+        $seen['outcome'] = get_class($e) . ':' . $e->getCode();
+        $seen['writable_after'] = $res->isWritable();
+    }
+});
+
+$cli = spawn(function () use ($port, $server) {
+    usleep(50000);
+
+    $fp = stream_socket_client("tcp://127.0.0.1:$port", $errno, $errstr, 5);
+    $sock = socket_import_stream($fp);
+    socket_set_option($sock, SOL_SOCKET, SO_RCVBUF, 4096);
+
+    fwrite($fp, "GET /stream HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n");
+
+    /* Take the opening frames, then stop reading so the handler parks. */
+    delay(200);
+    stream_set_blocking($fp, false);
+    $head = '';
+
+    for ($i = 0; $i < 40 && strlen($head) < 16384; $i++) {
+        $c = fread($fp, 8192);
+        if ($c !== false) $head .= $c;
+        usleep(5000);
+    }
+
+    delay(500);
+    $server->stop();
+    delay(500);
+
+    /* Whatever the cancelled response still had to say. */
+    $tail = '';
+
+    for ($i = 0; $i < 60; $i++) {
+        $c = fread($fp, 65536);
+        if ($c !== false) $tail .= $c;
+        usleep(5000);
+    }
+
+    fclose($fp);
+
+    [, $rest] = explode("\r\n\r\n", $head, 2);
+    $sizes = [];
+    $off   = 0;
+
+    while (count($sizes) < 3) {
+        $eol = strpos($rest, "\r\n", $off);
+        if ($eol === false) break;
+        $len = hexdec(substr($rest, $off, $eol - $off));
+        $off = $eol + 2;
+        if ($len === 0 || $off + $len + 2 > strlen($rest)) break;
+        /* The frame must be exactly what the handler wrote, and end where the
+         * size line said it would — a released or misplaced slot shows here. */
+        if (substr($rest, $off, $len) !== str_repeat('x', $len)) break;
+        if (substr($rest, $off + $len, 2) !== "\r\n") break;
+        $sizes[] = $len;
+        $off += $len + 2;
+    }
+
+    echo "frames=", implode(',', $sizes), "\n";
+    echo "terminator_after_cancel=", (int)(strpos($tail, "0\r\n\r\n") !== false), "\n";
+});
+
+$server->start();
+await($cli);
+
+foreach ($seen as $k => $v) echo "$k = ", var_export($v, true), "\n";
+echo "done\n";
+?>
+--EXPECT--
+frames=2000,2000,2000
+terminator_after_cancel=0
+outcome = 'TrueAsync\\HttpException:499'
+writable_after = false
+done