fix(core): unpublish a pool worker's inbox on every stop (#346)

CHANGELOG.md

@@ -19,6 +19,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **Under the reactor pool a worker stopped by its own handler kept taking HTTP/3 requests, and ran them after the stop (#346).** Only a stop through the pool took the worker's request inbox out of the registry the reactors pick from. `$server->stop()` in a handler stopped that worker's clone and left the inbox published, so the reactors routed requests to it while `start()` drained its scope, and each ran with `isRunning()` false: 6 to 10 of 16 requests after the stop (`h3/079`) and 6 to 7 of 16 sent while the stopping handler still held its thread (`h3/080`), 3 runs of 3 each. `stop()` now unpublishes the inbox at once, and `start()` waits for the reactors to finish any dispatch already under way before it drains. A handler's `stop()` during a stop through the pool no longer lets `start()` return while the pool's retire still waits on the reactors, and a worker that leaves by a fatal error unpublishes its inbox before it is freed.
 - **On Windows an upload failed with error 6 (`UPLOAD_ERR_NO_TMP_DIR`) when uploads ran in parallel, and a moved or kept upload could cost another upload its file (#368).** The end of a request deleted each temporary file by path, and the `UploadedFile` holding it deleted the same path again when it was released, by which time the file could be gone or moved and the name taken by a new upload. Windows reuses those names at once: `GetTempFileName` keeps only `tru` of the prefix and numbers the rest from the clock, so the second delete removed a file another process had just created. `multipart/009` failed 1 of 4 CI runs of a branch; its body, run 400 times across 8 processes, failed 6 to 10 times, and 0 of 1200 after the fix. The `UploadedFile` owns its file and deletes it once, so a file kept past its request lives until the object is released; a file `fopen` cannot open is deleted at once instead of left in the temporary directory.
 - **Over HTTP/1 a file body that failed or came up short left the connection open, and the next response was read as the rest of the body (#362).** The static handler and `sendFile()` ignored the engine's failure status, and plaintext sendfile reported a transfer that stopped at EOF as complete; a file truncated from 2 MiB to 4 KiB under the open-file cache sent 4,096 bytes and then the `200` of the request pipelined behind it, plaintext and TLS alike, 5 runs of 5. The connection now closes after such a body, and the cache entry that sized it is dropped. A body start the event loop refused used to complete inside the parser: the static handler answered the pipelined request `400` and `sendFile()` left its own unanswered; the request is now answered whole instead. With php-async 1521d69 (true-async/php-async#308) a plaintext body of 2 MiB also reaches a macOS client whole, where it was cut at about 1 MB, and Windows reports a short transfer instead of the full length. Evidence: `h1/065` to `h1/068`.
 - **Over HTTP/2 a file body that failed or came up short ended with END_STREAM, so the peer took the bytes it got for the whole file (#364).** A file truncated from 2 MiB to 100 KiB under the open-file cache sent 102,400 bytes and a clean end of stream, 5 runs of 5, and the cache entry that sized it stayed for its TTL. The stream now ends in RST_STREAM(INTERNAL_ERROR) after the bytes that were read, the connection serves the next request, and the entry is dropped. The short read also left its unread part charged to the per-worker static-file budget for good. Evidence: `h2/071`, `h2/072`.

dev/PLAN.md

@@ -205,8 +205,24 @@ an item marked "reproduce" gets its failing run before any code.
     and the macOS and Windows CI legs over the coming runs. Port 0 listeners
     (81c0437) stay the stronger form for tests that can read
     `getBoundListeners()`.
-15. #346 — the reactor-pool worker inbox spawns into the server scope after
-    `stop()` and across a restart, the path #345 left out: reproduce.
+15. #346 — the reactor-pool worker inbox after `stop()`: done. A handler's
+    `stop()` on its worker clone left the inbox in the registry; on main
+    `h3/079` reads 6 to 10 of 16 requests after the stop answered by the
+    stopped clone, `h3/080` 6 to 7 of 16 sent while the stopping handler held
+    its thread, 3 runs of 3 each, and both pass with the fix. `do_stop`
+    unpublishes the inbox; `start()` fences the reactors after its wait and
+    waits out a retire the pool coroutine still runs; the bailout and free
+    paths unpublish before the free. Mutants: dropping the `do_stop` unpublish
+    fails `h3/080`; dropping the fence in `start()` passes every local pool
+    test (a dispatch that loaded the inbox just before the unpublish, no hook
+    widens it). The restart half is unreachable: the pool loads a fresh clone
+    per worker run and starts it once.
+15a. #376 — `start()` from a handler after `stop()` runs inside the outer
+    run: it releases the scope the outer `start()` drains next
+    (`http_server_class.c`, the restart branch before the scope is created),
+    then fails at the room hub attach with "all 1024 slots are taken" while
+    one slot is held. Settle refuse-or-wait in the issue, then a test that
+    fails first.
 16. #348 — static passthrough on HTTP/1: a file under `on_missing: Next` that
     fails the engine's `open()` (mode 0) started the handler twice on one ctx
     and crashed, 3 of 3; PR 349 drops the HTTP/1 hook that spawned it.

src/http_server_class.c

@@ -372,6 +372,13 @@ static void http_server_listener_release(http_listener_t *listener)
 
 typedef struct pool_ctl_s pool_ctl_t;   /* pool control channel (#117), defined below */
 
+/* How far the retire of a worker clone's request inbox has got. */
+typedef enum {
+    WORKER_INBOX_LIVE = 0,
+    WORKER_INBOX_RETIRING,   /* unpublished; the fence or the drain is in flight */
+    WORKER_INBOX_RETIRED,    /* no reactor can post into it any more */
+} worker_inbox_state_t;
+
 /* Server object structure.
  * Field order grouped by alignment to minimise padding. zend_object must
  * remain last per PHP object layout contract.
@@ -447,6 +454,11 @@ struct http_server_object {
      * here and the drain dispatches them on this thread. */
     worker_inbox_t          *worker_inbox;
 
+    /* A stop through the pool retires the inbox on its own coroutine, and
+     * start() retires it after its wait; the second retire waits for the
+     * first instead of running again. */
+    worker_inbox_state_t     worker_inbox_state;
+
 #ifdef HAVE_HTTP_SERVER_HTTP3
     /* HTTP/3 UDP listeners — parallel to TCP listeners[] because they have
      * different transport semantics (no accept(), no per-connection fd) and
@@ -793,6 +805,7 @@ static void http_server_hot_reload_up(http_server_object *server, zval *this_zv,
                                       http_server_config_t *cfg);
 static void http_server_hot_reload_down(http_server_object *server);
 static void http_server_worker_inbox_retire(http_server_object *server);
+static void http_server_worker_inbox_unpublish(http_server_object *server);
 
 /* Runs on a fresh coroutine, not on the control wakeup's own callback: the stop
  * below disposes libuv handles, which must not happen from inside a handle's
@@ -813,6 +826,11 @@ static void http_server_pool_retire_entry(void)
      * stop — a producer must never post into a dying worker's inbox (#93). */
     http_server_worker_inbox_retire(server);
 
+    /* The retire suspends, and a handler's stop() may have stopped us meanwhile. */
+    if (!server->running || server->stopping) {
+        return;
+    }
+
     server->stopping = true;
     http_server_do_stop(server, reason);   /* logs server.stop reason=... */
 }
@@ -3444,13 +3462,26 @@ static void http_server_inbox_fence_cb(void *arg)
  * producers, which is the contract it always assumed. */
 static void http_server_worker_inbox_retire(http_server_object *server)
 {
-    if (server->worker_inbox == NULL || g_worker_registry == NULL) {
+    if (server->worker_inbox == NULL || g_worker_registry == NULL
+        || server->worker_inbox_state == WORKER_INBOX_RETIRED) {
         return;
     }
 
-    worker_registry_retire(g_worker_registry, server->worker_inbox);
-
     zend_coroutine_t *const co = ZEND_ASYNC_CURRENT_COROUTINE;
+
+    /* start() woken by a handler's stop() while the pool's retire is still
+     * fencing: returning now would let start() free the server under it. */
+    if (server->worker_inbox_state == WORKER_INBOX_RETIRING) {
+        while (co != NULL && server->worker_inbox_state == WORKER_INBOX_RETIRING) {
+            hot_reload_sleep_ms(co, 5);
+        }
+
+        return;
+    }
+
+    http_server_worker_inbox_unpublish(server);
+    server->worker_inbox_state = WORKER_INBOX_RETIRING;
+
     const int n = g_reactor_pool != NULL ? reactor_pool_count(g_reactor_pool) : 0;
 
     if (n > 0 && co != NULL) {
@@ -3502,6 +3533,18 @@ static void http_server_worker_inbox_retire(http_server_object *server)
 
     http_logf_info(&server->log_state, "reload.inbox retired depth=%zu",
                    worker_inbox_depth(server->worker_inbox));
+    server->worker_inbox_state = WORKER_INBOX_RETIRED;
+}
+
+/* Step 1 of the retire alone: no reactor picks this worker from now on. It
+ * does not suspend, so do_stop runs it on every stop, whichever way the stop
+ * came; without it a clone stopped by its own handler takes requests until
+ * start() resumes, and runs them after the stop. Idempotent. */
+static void http_server_worker_inbox_unpublish(http_server_object *server)
+{
+    if (server->worker_inbox != NULL && g_worker_registry != NULL) {
+        (void)worker_registry_retire(g_worker_registry, server->worker_inbox);
+    }
 }
 
 /* Claim this worker clone's stats slab slot and point its live counters at it,
@@ -4292,10 +4335,6 @@ ZEND_METHOD(TrueAsync_HttpServer, start)
      * explicitly release it. */
     server->scope_object = server->server_scope->scope_object;
 
-    /* Worker-pool clone under the reactor-pool gate: publish a request inbox so
-     * a reactor can route parsed requests to this worker. No-op otherwise. */
-    http_server_worker_inbox_up(server);
-
     /* Build TLS context up-front if any listener declared tls=true.
      * Doing this *before* binding sockets keeps the failure path cheap:
      * a bad cert means no listen_event allocation at all, and the
@@ -4751,6 +4790,14 @@ ZEND_METHOD(TrueAsync_HttpServer, start)
         RETURN_FALSE;
     }
 
+    /* Worker-pool clone under the reactor-pool gate: publish a request inbox so
+     * a reactor can route parsed requests to this worker. No-op otherwise.
+     * Not when a stop() already waits: the inbox would take requests until the
+     * do_stop below unpublished it, and run them after the stop. */
+    if (!server->stopping) {
+        http_server_worker_inbox_up(server);
+    }
+
     bool bailout = false;
 
     if (UNEXPECTED(server->stopping)) {
@@ -4762,6 +4809,8 @@ ZEND_METHOD(TrueAsync_HttpServer, start)
     } else if (!http_server_await_stop(server, &bailout)) {
         /* This path never reaches the exits below, so the slot and the hub
          * reference are dropped here instead. */
+        http_server_worker_inbox_unpublish(server);
+
         if (room_hub_attached) {
             room_hub_detach(server->topic_hub);
         }
@@ -4786,6 +4835,9 @@ ZEND_METHOD(TrueAsync_HttpServer, start)
             room_hub_detach_request_over(server->topic_hub);
         }
 
+        /* No fence on this path: unpublishing at least keeps new picks off an
+         * inbox that http_server_free frees. */
+        http_server_worker_inbox_unpublish(server);
         zend_bailout();
     }
 
@@ -4795,6 +4847,11 @@ ZEND_METHOD(TrueAsync_HttpServer, start)
         http_server_do_stop(server, "shutdown");
     }
 
+    /* do_stop unpublished the inbox but cannot suspend for the fence; a stop
+     * through the pool has run the whole retire before it already. The
+     * requests still in the inbox drain into server_scope with the rest. */
+    http_server_worker_inbox_retire(server);
+
     /* Drain in-flight per-request handler coroutines now, while we are still
      * on the start() coroutine and can suspend, so server_scope is empty when
      * the object is freed (issue #74). On bailout we already longjmp'd above. */
@@ -4869,6 +4926,7 @@ static void http_server_do_stop(http_server_object *server, const char *reason)
 #endif
 
     http_server_retire_connections(server);
+    http_server_worker_inbox_unpublish(server);
 
     /* Logger teardown must precede the wait_event notify: waking
      * start() can let the scheduler tear down before our async write
@@ -6363,8 +6421,11 @@ static void http_server_free(zend_object *obj)
     http_server_reactor_pool_down(server);
 
     /* This worker clone's request inbox. Producers (reactors) have quiesced by
-     * the time a worker is freed. */
+     * the time a worker is freed, unless start() left by a bailout or a failed
+     * wait: those paths do not fence, and only the unpublish keeps a reactor
+     * from picking an inbox freed here. */
     if (server->worker_inbox != NULL) {
+        http_server_worker_inbox_unpublish(server);
         worker_inbox_free(server->worker_inbox);
         server->worker_inbox = NULL;
     }

tests/phpt/server/h3/079-h3-pool-worker-stops-itself.phpt

@@ -0,0 +1,100 @@
+--TEST--
+HttpServer: a reactor-pool worker stopped by its own handler receives no more requests (#346)
+--EXTENSIONS--
+true_async_server
+true_async
+--SKIPIF--
+<?php
+require __DIR__ . '/_h3_skipif.inc';
+if (PHP_OS_FAMILY === 'Windows') die('skip libuv on Windows lacks SO_REUSEPORT');
+h3_skipif(['openssl_cli' => true, 'h3client' => true]);
+?>
+--ENV--
+TRUE_ASYNC_SERVER_REACTOR_POOL=1
+PHP_HTTP3_DISABLE_RETRY=1
+--FILE--
+<?php
+/* A handler on a pool worker holds that worker's clone of the server, and
+ * stop() on it stops the clone alone. The reactors route requests to the
+ * inboxes in the worker registry, so the clone's inbox leaves the registry
+ * with the stop: a clone still published takes its reactor's share of the
+ * requests after the stop and runs them with isRunning() === false. Every
+ * request after the stop has to reach the worker that still runs. */
+
+use TrueAsync\HttpServer;
+use TrueAsync\HttpServerConfig;
+use function Async\spawn;
+use function Async\delay;
+
+require __DIR__ . '/_h3_skipif.inc';
+require_once __DIR__ . '/../_free_port.inc';
+
+$tmp = __DIR__ . '/tmp-079';
+@mkdir($tmp, 0700, true);
+$cert = "$tmp/cert.pem";
+$key  = "$tmp/key.pem";
+if (!h3_gen_cert($key, $cert)) { echo "cert gen failed\n"; exit(1); }
+register_shutdown_function(function () use ($tmp, $cert, $key) {
+    @unlink("$tmp/stderr.txt"); @unlink($cert); @unlink($key); @rmdir($tmp);
+});
+
+$port = tas_free_port_span(2);
+$config = (new HttpServerConfig())
+    ->addListener('127.0.0.1', $port + 1)   /* TCP listener required by start() */
+    ->addHttp3Listener('127.0.0.1', $port)
+    ->enableTls(true)->setCertificate($cert)->setPrivateKey($key)
+    ->setWorkers(2);
+$server = new HttpServer($config);
+
+$server->addHttpHandler(function ($req, $res) use ($server) {
+    if ($req->getPath() === '/stop') {
+        $res->setBody('stopping');
+        $server->stop();
+        return;
+    }
+
+    $res->setBody($server->isRunning() ? 'running' : 'STOPPED');
+});
+
+$client_bin = __DIR__ . '/../../../h3client/h3client';
+
+function h3_get(string $bin, int $port, string $path, string $tmp): string
+{
+    $cmd = sprintf('H3CLIENT_DEADLINE_MS=4000 %s 127.0.0.1 %d %s GET 2>%s',
+        escapeshellarg($bin), $port, escapeshellarg($path), escapeshellarg("$tmp/stderr.txt"));
+    $body = (string) shell_exec($cmd);
+    preg_match('/^STATUS=(\d+)$/m', (string) @file_get_contents("$tmp/stderr.txt"), $m);
+
+    return ($m[1] ?? '-') . ' ' . $body;
+}
+
+spawn(function () use ($server, $port, $client_bin, $tmp) {
+    /* Reactors + workers need a moment to thread up and bind. */
+    delay(600);
+
+    /* Taken before the echo: the stopped worker's exit notice goes to stderr
+     * while the request is in flight. */
+    $stop = h3_get($client_bin, $port, '/stop', $tmp);
+    echo "/stop: $stop\n";
+
+    $answers = [];
+    for ($i = 0; $i < 16; $i++) {
+        $answer = h3_get($client_bin, $port, '/', $tmp);
+        $answers[$answer] = ($answers[$answer] ?? 0) + 1;
+    }
+
+    ksort($answers);
+    foreach ($answers as $answer => $count) {
+        echo "after: $answer x$count\n";
+    }
+
+    $server->stop();
+});
+
+$server->start();
+echo "done\n";
+?>
+--EXPECTF--
+%A/stop: 200 stopping
+%Aafter: 200 running x16
+%Adone