fix(grpc/http): review follow-ups — mask drift, body caps, flow-control, HEAD, compression API

CHANGELOG.md

@@ -25,9 +25,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
     handler that writes no messages gets the canonical Trailers-Only reply.
   - **grpc-web (binary)**: `application/grpc-web` calls carry their trailers
     in-body as a `0x80`-flagged frame, on H2 and H3.
-  - **Per-message gzip**: inbound `grpc-encoding: gzip` messages inflate
-    transparently in `readMessage()`; `writeMessage(..., compress: true)`
-    emits compressed frames.
+  - **gzip message encoding**: inbound `grpc-encoding: gzip` messages inflate
+    transparently in `readMessage()`; `setGrpcEncoding('gzip')` declared
+    before the first `writeMessage()` compresses every reply frame (per-call
+    declaration, mirroring grpc-go/java/C++ — a compressed frame without a
+    declared encoding cannot be expressed).
   - **`grpc-timeout`** request header parsed and exposed via
     `HttpRequest::getGrpcTimeout()`.
   - **grpc-web-text**: `application/grpc-web-text` calls carry base64 both

README.md

@@ -308,8 +308,9 @@ Unary and all three streaming shapes use this one API; `grpc-status` /
 `grpc-message` ride real HTTP trailers (or the in-body `0x80` frame for
 grpc-web, base64-encoded per frame for grpc-web-text), uncaught exceptions
 map to `13 INTERNAL`, inbound `grpc-encoding: gzip` messages inflate
-transparently, and `writeMessage($msg, compress: true)` gzips replies. The
-client's deadline is exposed via `$request->getGrpcTimeout()`.
+transparently, and `setGrpcEncoding('gzip')` before the first message gzips
+every reply frame. The client's deadline is exposed via
+`$request->getGrpcTimeout()`.
 
 Working examples live under [`examples/`](examples/):
 [`minimal-server.php`](examples/minimal-server.php),

include/core/async_plain_event.h

@@ -21,4 +21,9 @@ static zend_always_inline void async_plain_event_fire(zend_async_event_t *event)
     }
 }
 
+/* Suspend `co` for `ms` on a one-shot timer; the worker loop keeps draining
+ * (mailbox, scheduler) meanwhile. Leaves a resume exception in EG for the
+ * caller to inspect or clear (a create failure is cleared internally). */
+void async_coroutine_sleep_ms(zend_coroutine_t *co, zend_ulong ms);
+
 #endif /* ASYNC_PLAIN_EVENT_H */

include/core/stream_credit.h

@@ -14,6 +14,7 @@
 #include <stdbool.h>
 
 #include "Zend/zend_atomic.h"
+#include "Zend/zend_async_API.h"   /* zend_async_trigger_event_t */
 
 /* Per-stream flow-control credit shared worker↔reactor. Malloc + atomics
  * only — the one object both threads touch. Worker creates it with two refs
@@ -24,6 +25,8 @@ typedef struct {
     zend_atomic_int64 acked;   /* bytes the reactor has retired (peer ACK) */
     zend_atomic_int   dead;    /* stream died — producer must stop waiting */
     zend_atomic_int   refs;
+    zend_atomic_ptr   waker;      /* worker-owned trigger the reactor signals */
+    zend_atomic_int   waker_busy; /* a signaller is inside trigger() */
 } stream_credit_t;
 
 static inline stream_credit_t *stream_credit_create(void)
@@ -37,6 +40,8 @@ static inline stream_credit_t *stream_credit_create(void)
     ZEND_ATOMIC_INT64_INIT(&sc->acked, 0);
     ZEND_ATOMIC_INT_INIT(&sc->dead, 0);
     ZEND_ATOMIC_INT_INIT(&sc->refs, 2);   /* worker ctx + reactor stream */
+    ZEND_ATOMIC_PTR_INIT(&sc->waker, NULL);
+    ZEND_ATOMIC_INT_INIT(&sc->waker_busy, 0);
 
     return sc;
 }
@@ -66,11 +71,46 @@ static inline void stream_credit_mark_dead(stream_credit_t *sc)
     zend_atomic_int_store_ex(&sc->dead, 1);
 }
 
-/* NULL-safe. Dead must be set before release so a parked producer sees it. */
+/* Signal the parked producer from any thread. busy brackets the load+trigger
+ * so stream_credit_clear_waker can't dispose the event mid-signal. */
+static inline void stream_credit_wake(stream_credit_t *sc)
+{
+    zend_atomic_int_store_ex(&sc->waker_busy, 1);
+
+    zend_async_trigger_event_t *const t =
+        (zend_async_trigger_event_t *)zend_atomic_ptr_load_ex(&sc->waker);
+
+    if (t != NULL) {
+        t->trigger(t);
+    }
+
+    zend_atomic_int_store_ex(&sc->waker_busy, 0);
+}
+
+/* Worker thread: publish/retract the park trigger. clear must complete
+ * before the worker disposes the event. */
+static inline void stream_credit_set_waker(stream_credit_t *sc,
+                                           zend_async_trigger_event_t *t)
+{
+    zend_atomic_ptr_store_ex(&sc->waker, t);
+}
+
+static inline void stream_credit_clear_waker(stream_credit_t *sc)
+{
+    zend_atomic_ptr_store_ex(&sc->waker, NULL);
+
+    /* wait out a signal that loaded the pointer just before the NULL store */
+    while (zend_atomic_int_load_ex(&sc->waker_busy) != 0) {
+    }
+}
+
+/* NULL-safe. Dead must be set before release so a parked producer sees it;
+ * wake so it sees it NOW instead of at its next timeout. */
 static inline void stream_credit_abandon(stream_credit_t *sc)
 {
     if (sc != NULL) {
         stream_credit_mark_dead(sc);
+        stream_credit_wake(sc);
         stream_credit_release(sc);
     }
 }

include/core/worker_dispatch.h

@@ -38,6 +38,12 @@ typedef struct http_server_object http_server_object;
  * the stream (STREAM_* fragments must never be dropped silently). */
 typedef bool (*worker_response_sink_fn)(response_wire_t *rw, void *sink_arg);
 
+/* Discard an undeliverable wire: abandon its credit (marks the producer's
+ * stream dead), release the owned chunk, free the wire. Zend-side companion
+ * to response_wire_free — every drop site must go through this so a new
+ * owned field can't leak from a forgotten copy. */
+void response_wire_discard(response_wire_t *rw);
+
 /* Take ownership of `req` (a persistent reactor-built or ZMM request, refcount
  * 1), wrap it in an HttpRequest on THIS (worker) thread, spawn the user handler
  * coroutine in `scope`, and when it finishes render the HttpResponse into a

include/http1/http_parser.h

@@ -215,6 +215,13 @@ struct http_request_t {
     void                     *body_h3_conn;
     int64_t                   body_h3_stream_id;
     size_t                    body_h3_uncredited;  /* drained, not yet flushed */
+
+    /* Bytes spliced into the queue by a buffered→streaming upgrade whose
+     * flow-control credit was already returned while they sat in the
+     * transport's buffer (H2 buffered consume / H3 immediate extend).
+     * http_body_stream_pop drains this before granting new credit so the
+     * same bytes are never credited twice. */
+    size_t                    body_precredited;
     int32_t                   body_h2_consume_pending;
     void                     *body_h3_stream;
 
@@ -235,6 +242,10 @@ struct http_request_t {
     bool         body_streaming;
     bool         body_eof;
     bool         body_error;
+
+    /* grpc_mode_t stamped once at headers-complete; body policy derives
+     * from it (http_request_body_must_buffer / _size_uncapped). */
+    uint8_t      grpc_mode;
     /* TODO(issue #26 backpressure): set true when on_body trips
      * llhttp_pause; readBody clears it via llhttp_resume below the
      * low-water mark. Unused by the MVP — see TODO in on_body. */

include/php_http_server.h

@@ -1126,6 +1126,10 @@ zend_string *http_response_format_static_head(zend_object *obj,
                                               bool include_inline_body);
 void http_response_set_socket(zend_object *obj, php_socket_t fd);
 void http_response_set_protocol_version(zend_object *obj, const char *version);
+/* RFC 9110 §9.3.2 — HEAD responses must not carry a body; send() drops
+ * chunks silently when set. Stamped at dispatch wherever the request is
+ * known. */
+void http_response_set_head(zend_object *obj, bool is_head);
 bool http_response_is_closed(zend_object *obj);
 
 /* HTTP/2 strategy uses these to build frames without going through

src/core/async_plain_event.c

@@ -11,6 +11,7 @@
 #endif
 
 #include "php.h"
+#include "Zend/zend_exceptions.h"   /* zend_clear_exception */
 #include "Zend/zend_async_API.h"
 #include "core/async_plain_event.h"
 
@@ -52,6 +53,21 @@ static bool plain_dispose(zend_async_event_t *event)
     return true;
 }
 
+void async_coroutine_sleep_ms(zend_coroutine_t *co, const zend_ulong ms)
+{
+    zend_async_timer_event_t *const t = ZEND_ASYNC_NEW_TIMER_EVENT(ms, false);
+
+    if (UNEXPECTED(t == NULL)) {
+        zend_clear_exception();
+        return;
+    }
+
+    t->base.start(&t->base);
+    zend_async_resume_when(co, &t->base, true, zend_async_waker_callback_resolve, NULL);
+    ZEND_ASYNC_SUSPEND();
+    ZEND_ASYNC_WAKER_DESTROY(co);
+}
+
 zend_async_event_t *async_plain_event_new(void)
 {
     zend_async_event_t *event = pecalloc(1, sizeof(*event), 0);

src/core/http_connection.c

@@ -2262,6 +2262,8 @@ static void http_connection_dispatch_request(http_connection_t *conn, http_reque
     /* Create HttpResponse PHP object */
     object_init_ex(&ctx->response_zv, http_response_ce);
     http_response_set_protocol_version(Z_OBJ(ctx->response_zv), conn->http_version);
+    http_response_set_head(Z_OBJ(ctx->response_zv),
+                           http_request_method_is_head(req));
 
     /* Install the HTTP/1 streaming vtable. send() on the response
      * activates chunked framing on first call; handlers that stick to

src/core/http_protocol_handlers.c

@@ -12,6 +12,8 @@
 
 #include "php.h"
 #include "http_protocol_handlers.h"
+#include "http1/http_parser.h"   /* http_request_t */
+#include "grpc/grpc.h"           /* grpc_request_mode */
 
 /* {{{ http_protocol_type_to_string */
 const char* http_protocol_type_to_string(http_protocol_type_t type)
@@ -127,6 +129,23 @@ bool http_protocol_has_handler(HashTable *handlers, http_protocol_type_t protoco
 }
 /* }}} */
 
+/* {{{ http_request_classify_protocols */
+void http_request_classify_protocols(http_request_t *req)
+{
+    req->grpc_mode = (uint8_t)grpc_request_mode(req);
+}
+/* }}} */
+
+bool http_request_body_must_buffer(const http_request_t *req)
+{
+    return req->grpc_mode == GRPC_MODE_WEB_TEXT;
+}
+
+bool http_request_body_size_uncapped(const http_request_t *req)
+{
+    return req->grpc_mode != GRPC_MODE_NONE;
+}
+
 /* {{{ http_protocol_pick_handler */
 zend_fcall_t *http_protocol_pick_handler(HashTable *handlers, const bool is_grpc)
 {

src/core/http_protocol_handlers.h

@@ -35,6 +35,18 @@ bool http_protocol_has_handler(HashTable *handlers, http_protocol_type_t protoco
  * classified) → HTTP1 → HTTP2. NULL when nothing matches. */
 zend_fcall_t *http_protocol_pick_handler(HashTable *handlers, bool is_grpc);
 
+/* Stamp the request's grpc_mode once at headers-complete, before any
+ * body-streaming decision; transports never classify themselves and read
+ * the body policy through the predicates below. */
+void http_request_classify_protocols(struct http_request_t *req);
+
+/* Body policy derived from the stamped grpc_mode, keeping transports
+ * gRPC-agnostic. must_buffer: never stream (grpc-web-text decodes the whole
+ * body). size_uncapped: waive the cumulative max_body_size cap (gRPC streams
+ * are unbounded; the in-flight window still bounds memory). */
+bool http_request_body_must_buffer(const struct http_request_t *req);
+bool http_request_body_size_uncapped(const struct http_request_t *req);
+
 /* Remove handler from HashTable */
 void http_protocol_remove_handler(HashTable *handlers, http_protocol_type_t protocol);
 

src/core/http_protocol_strategy.h

@@ -129,6 +129,25 @@ typedef uint32_t http_protocol_mask_t;
                                   HTTP_PROTO_MASK_SSE   | HTTP_PROTO_MASK_GRPC)
 #define HTTP_PROTO_MASK_HAS(mask, type) (((mask) & (1u << (type))) != 0)
 
+/*
+ * Mask bits a registered handler of `type` enables. The single source for
+ * the protocol→mask rule — used both by the addXHandler methods at
+ * registration and by the worker-side transit rebuild, so the two can't
+ * drift (a plain HTTP handler serves h1+h2 on one port; gRPC rides h2).
+ */
+static zend_always_inline http_protocol_mask_t
+http_protocol_registration_mask(http_protocol_type_t type)
+{
+    switch (type) {
+        case HTTP_PROTOCOL_HTTP1:
+            return HTTP_PROTO_MASK_HTTP1 | HTTP_PROTO_MASK_HTTP2;
+        case HTTP_PROTOCOL_GRPC:
+            return HTTP_PROTO_MASK_GRPC | HTTP_PROTO_MASK_HTTP2;
+        default:
+            return (http_protocol_mask_t)(1u << type);
+    }
+}
+
 /*
  * Sentinel returned internally when the byte prefix matches no
  * protocol the server accepts (e.g. HTTP/2 preface on a listener with

src/core/reactor_pool.c

@@ -281,6 +281,13 @@ bool reactor_pool_exec(reactor_pool_t *rp, const int idx, const reactor_exec_fn
 
     /* Acquire: pair with the reactor's release store once fn has run. */
     while (zend_atomic_int_load_ex(&done) == 0) {
+        if (zend_atomic_int_load_ex(&rc->phase) == REACTOR_PHASE_DONE) {
+            /* Loop exited: mailbox_free discards still-queued cmds (they can
+             * never run — no stack UAF), it runs before the DONE store. One
+             * final check catches a cmd that executed in the last drain. */
+            return zend_atomic_int_load_ex(&done) != 0;
+        }
+
         reactor_pool_msleep();
     }