<?php
/**
* SeekQuarry/Yioop --
* Open Source Pure PHP Search Engine, Crawler, and Indexer
*
* Copyright (C) 2009 - 2026 Chris Pollett chris@pollett.org
*
* LICENSE:
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see
* <https://www.gnu.org/licenses/>.
*
* END LICENSE
*
* @author Chris Pollett chris@pollett.org
* @license https://www.gnu.org/licenses/ GPL3
* @link https://www.seekquarry.com/
* @copyright 2009 - 2026
* @filesource
*/
namespace seekquarry\yioop\tests;
use seekquarry\atto\Connection;
use seekquarry\atto\WebSocket;
use seekquarry\atto\WebSite;
use seekquarry\atto\WindowUpdateFrame;
use seekquarry\yioop\library\UnitTest;
/**
* A WebSite that opens up the few protected HTTP/2 send-path pieces a
* test needs to drive a response by hand: setting up the per-stream
* flow-control windows, starting a paced response body, feeding one
* received frame in, draining whatever is queued to the client, and
* registering a fake connection so the drain can find it.
*/
class H2PacedResumeProbeWebSite extends WebSite
{
/**
* Sets up the connection's HTTP/2 send windows the way the server
* does right after the client's opening SETTINGS frame is read.
*
* @param object $conn connection holding the protocol state
* @param array $client_settings client settings, keyed by the
* HTTP/2 settings ids, used here to seed the per-stream
* initial window
* @return void
*/
public function initStreamWindows($conn, $client_settings)
{
$this->initH2SendWindows($conn, $client_settings);
}
/**
* Starts sending a response body on a stream, pacing it across the
* event loop when it is larger than the stream's send window.
*
* @param int $key connection identifier
* @param resource $socket socket the frames go out on
* @param object $conn connection holding the protocol state
* @param int $stream_id stream the body belongs to
* @param string $body the complete response body
* @return void
*/
public function startPacedBody($key, $socket, $conn, $stream_id,
$body)
{
$this->sendH2BodyPaced($key, $socket, $conn, $stream_id, $body);
}
/**
* Hands one received HTTP/2 frame to the server's frame handler,
* standing in for the event loop reading it off the wire.
*
* @param int $key connection identifier
* @param resource $socket socket the connection is on
* @param object $conn connection holding the protocol state
* @param object $frame the frame object to process
* @param string $payload the frame's raw body bytes
* @param int $length the frame body length in bytes
* @return mixed whatever the frame handler returns
*/
public function handleFrame($key, $socket, $conn, $frame, $payload,
$length)
{
return $this->processH2Frame($key, $socket, $conn, $frame,
$payload, $length);
}
/**
* Drains the connections that currently have unsent response bytes,
* writing as much as each socket will take, the way the event loop
* does once per pass.
*
* @return void
*/
public function drainResponses()
{
$this->processResponseStreams(
$this->out_streams[self::CONNECTION] ?? []);
}
/**
* Registers a connection and its socket so the drain step can find
* it, the way accepting a connection would.
*
* @param int $key connection identifier
* @param object $conn the connection object
* @param resource $socket the connection's socket
* @return void
*/
public function registerConnection($key, $conn, $socket)
{
$this->connections[$key] = $conn;
$this->in_streams[self::CONNECTION][$key] = $socket;
}
/**
* Reports whether the connection still has response bytes queued
* and not yet written to the client.
*
* @param int $key connection identifier
* @return bool true when bytes are still waiting to go out
*/
public function hasQueuedResponse($key)
{
return isset($this->out_streams[self::CONNECTION][$key])
&& isset($this->out_streams[self::DATA][$key]);
}
/**
* Reads and processes whatever has arrived on the connection's
* socket, the way the event loop does for a readable connection.
* This drives the real read path, including its handling of an
* end-of-file seen on the socket, rather than handing a frame
* straight to the frame handler.
*
* @param int $key connection identifier
* @param resource $socket the connection's socket to read from
* @return mixed whatever the read path returns
*/
public function readOnce($key, $socket)
{
return $this->parseH2Request($key, $socket);
}
}
/**
* A WebSite that records every frame the HTTP/2 read loop hands to the
* frame handler and reports the first one as a dispatched request, so a
* test can confirm the read loop keeps draining the buffer past that
* first dispatch instead of stopping there.
*/
class H2FrameDrainProbeWebSite extends WebSite
{
/**
* The stream ids of the frames the read loop handed off, in order,
* one entry per call.
* @var array
*/
public $seen = [];
/**
* Reads and processes whatever has arrived on the connection's
* socket, the way the event loop does for a readable connection.
*
* @param int $key connection identifier
* @param resource $socket the connection's socket to read from
* @return mixed whatever the read path returns
*/
public function readOnce($key, $socket)
{
return $this->parseH2Request($key, $socket);
}
/**
* Stands in for the real frame handler: records the frame's stream
* id and reports the very first frame as a dispatched request (a
* non-false return) and the rest as routine. The read loop must
* keep going after that first non-false return, so every frame in a
* single read shows up here.
*
* @param int $key connection identifier
* @param resource $connection the connection's socket
* @param object $conn connection holding the protocol state
* @param object $frame the parsed frame
* @param string $payload the frame's raw body bytes
* @param int $length the frame body length in bytes
* @return mixed true for the first frame, false afterward
*/
protected function processH2Frame($key, $connection, $conn, $frame,
$payload, $length)
{
$this->seen[] = $frame->stream_id;
return count($this->seen) === 1;
}
/**
* Registers a connection and its socket so the read path can find
* it, the way accepting a connection would.
*
* @param int $key connection identifier
* @param object $conn the connection object
* @param resource $socket the connection's socket
* @return void
*/
public function registerConnection($key, $conn, $socket)
{
$this->connections[$key] = $conn;
$this->in_streams[self::CONNECTION][$key] = $socket;
}
}
/**
* A WebSite that exposes the protected memory-sample timing check so a
* test can drive it with chosen clock values instead of the real time.
*/
class MemorySampleProbeWebSite extends WebSite
{
/**
* Sets the sample period the timing check reads, standing in for
* the value the running server takes from its config.
*
* @param int $seconds whole seconds between samples, zero for off
*/
public function setPeriod($seconds)
{
$this->memory_sample_period = $seconds;
}
/**
* Calls the protected timing check with a supplied current time.
*
* @param int $now pretend wall-clock time in whole seconds
* @return bool whether a memory sample is due at that time
*/
public function dueAt($now)
{
return $this->memorySampleDue($now);
}
}
/**
* Reaches the session storage of a WebSite so that writing, reading back
* and dropping a session can be checked without running a server.
*/
class SessionFileProbeWebSite extends WebSite
{
/**
* Points sessions at a directory of the caller's choosing and sets how
* large one may grow, standing in for what a running server takes from
* its configuration.
*
* @param string $directory where session files should be written
* @param int $largest how many bytes a written session may reach
*/
public function useDirectory($directory, $largest)
{
$this->default_server_globals['SESSION_DIR'] = $directory;
$this->default_server_globals['MAX_REQUEST_LEN'] = $largest;
}
/**
* Writes a session's contents through the storage under test.
*
* @param string $session_id the session to write
* @param array $session_data the contents to write
* @return bool whether the contents were written
*/
public function write($session_id, $session_data)
{
return $this->writeSessionData($session_id, $session_data);
}
/**
* Drops a session and the file its contents were kept in.
*
* @param string $session_id the session to drop
*/
public function forget($session_id)
{
$this->forgetSession($session_id);
}
/**
* The path a session's contents are kept at.
*
* @param string $session_id the session to name a file for
* @return string absolute path of that session's file
*/
public function pathOf($session_id)
{
return $this->sessionFilePath($session_id);
}
}
/**
* Gathers the tests for the atto WebSite server in one place. Some cover
* the HTTP/2 send path: a large response must be delivered in full even
* when it does not fit the client's opening per-stream window, the body
* is sent up to the window then parks at a zero send window, and the
* rest must go out once the client returns credit with a WINDOW_UPDATE
* frame, guarding against the partial-transfer failure where the tail
* was never sent; another checks every request coalesced into one read
* is dispatched, not just the first. The HTTP/2 cases run over an
* in-memory socket pair, so they are deterministic and touch no network.
* The rest cover the switchable memory-usage sampling: how often a
* sample is due based on the period the server is configured with.
*
* @author Chris Pollett
*/
class WebSiteTest extends UnitTest
{
/**
* How many streams the closed-stream case opens, completes, and
* then sends a late WINDOW_UPDATE for. Chosen larger than the
* HTTP/2 max-concurrent-streams guard so that, before the fix,
* the run would have stashed enough stranded credit entries to
* trip the guard and close the connection.
*/
const CLOSED_STREAM_RUN = 120;
/**
* The socket pair used by a case, closed in tearDown so a case that
* throws part way through still releases its sockets.
* @var array
*/
public $sockets;
/**
* Connection key the test writes against, standing for one client.
* @var int
*/
public $key = 7;
/**
* The server whose outbound framing is under test.
* @var WebSite
*/
public $site;
/**
* The streams a server reads from, reached through reflection
* because a server keeps them to itself. A write is dropped unless
* its connection is still among them.
* @var \ReflectionProperty
*/
public $in_streams;
/**
* The streams a server writes to, reached through reflection because
* a server keeps them to itself. What a test wants to read is the
* bytes waiting here.
* @var \ReflectionProperty
*/
public $out_streams;
/**
* The method that wraps socket frames in data frames, reached
* through reflection.
* @var \ReflectionMethod
*/
public $enqueue_h2;
/**
* The method that picks which way a socket frame is written,
* reached through reflection.
* @var \ReflectionMethod
*/
public $send_frame;
/**
* The method that says whether a drained connection carries nothing
* but a socket, reached through reflection.
* @var \ReflectionMethod
*/
public $uses_socket_drain;
/**
* Forces the atto WebSite file to load before a case builds the
* server and frame objects, since that file is self-contained and
* defines WebSite, Connection, and the frame classes together, then
* stands up a server and a connection for the cases that check how
* socket frames are written.
*/
public function setUp()
{
class_exists('seekquarry\\atto\\WebSite');
$this->sockets = [];
$reflection = new \ReflectionClass(WebSite::class);
$this->site = $reflection->newInstanceWithoutConstructor();
/* what a class keeps to itself has been reachable through
reflection without asking since PHP 8.1, and asking is an
error as of 8.5, so nothing is asked for here. */
$this->in_streams = new \ReflectionProperty(WebSite::class,
"in_streams");
$this->out_streams = new \ReflectionProperty(WebSite::class,
"out_streams");
$this->enqueue_h2 = new \ReflectionMethod(WebSite::class,
"wsEnqueueH2Write");
$this->send_frame = new \ReflectionMethod(WebSite::class,
"wsSendFrame");
$this->uses_socket_drain = new \ReflectionMethod(WebSite::class,
"usesSocketDrain");
$this->in_streams->setValue($this->site,
[WebSite::CONNECTION => [$this->key => "a-connection"]]);
$this->out_streams->setValue($this->site, []);
}
/**
* Closes the socket pair a case opened.
*/
public function tearDown()
{
foreach ($this->sockets as $socket) {
if (is_resource($socket)) {
fclose($socket);
}
}
$this->sockets = [];
}
/**
* Builds a connection on one end of an in-memory socket pair with
* its HTTP/2 send windows set up like a real client handshake: a
* per-stream initial window the body will not fit in, and a large
* connection-level window so the connection window is never the
* limit.
*
* @param object $site the probe WebSite driving the exchange
* @param int $stream_window the client's per-stream initial window
* @return array [connection key, connection object, server socket]
*/
private function setUpConnection($site, $stream_window)
{
$pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM,
0);
$server = $pair[0];
$client = $pair[1];
stream_set_blocking($server, false);
stream_set_blocking($client, false);
$this->sockets = [$server, $client];
$key = (int)$server;
$connection = new Connection($server, false);
$connection->protocol = 'h2';
$connection->is_websocket = false;
$site->initStreamWindows($connection,
[WebSite::H2_SETTING_INITIAL_WINDOW => $stream_window]);
$connection->protocol_state['send_window'] = 12436774;
$site->registerConnection($key, $connection, $server);
return [$key, $connection, $server];
}
/**
* Drains queued response bytes to the socket while reading the
* client end, so a full socket buffer never blocks further writes,
* and returns all bytes read from the client end so far appended to
* what was already collected.
*
* @param object $site the probe WebSite to drain
* @param int $key connection identifier to drain
* @param resource $client the client end to read from
* @param string $collected bytes read on earlier drains
* @return string $collected plus the newly read bytes
*/
private function drainAndRead($site, $key, $client, $collected)
{
for ($pass = 0; $pass < 2000
&& $site->hasQueuedResponse($key); $pass++) {
$site->drainResponses();
while (($chunk = fread($client, 65536)) !== false
&& $chunk !== "") {
$collected .= $chunk;
}
}
while (($chunk = fread($client, 65536)) !== false
&& $chunk !== "") {
$collected .= $chunk;
}
return $collected;
}
/**
* Sums the payload bytes of the DATA frames addressed to one stream
* inside a buffer of serialized HTTP/2 frames, and reports whether
* any of them carried the END_STREAM flag. Each frame is a nine
* byte header (three byte length, one byte type, one byte flags,
* four byte stream id) followed by its payload.
*
* @param string $buffer serialized frame bytes read from the client
* @param int $stream the stream id whose DATA payload to total
* @return array [total DATA payload bytes, whether END_STREAM seen]
*/
private function countDataPayload($buffer, $stream)
{
$position = 0;
$total = 0;
$length = strlen($buffer);
$end_seen = false;
while ($position + 9 <= $length) {
$frame_length = (ord($buffer[$position]) << 16)
| (ord($buffer[$position + 1]) << 8)
| ord($buffer[$position + 2]);
$type = ord($buffer[$position + 3]);
$flags = ord($buffer[$position + 4]);
$frame_stream = ((ord($buffer[$position + 5]) & 0x7f) << 24)
| (ord($buffer[$position + 6]) << 16)
| (ord($buffer[$position + 7]) << 8)
| ord($buffer[$position + 8]);
$position += 9;
if ($position + $frame_length > $length) {
break;
}
if ($type === 0x00 && $frame_stream === $stream) {
$total += $frame_length;
if ($flags & 0x01) {
$end_seen = true;
}
}
$position += $frame_length;
}
return [$total, $end_seen];
}
/**
* A response larger than the client's per-stream window is sent up
* to the window, parks, and then the rest goes out after a
* WINDOW_UPDATE returns credit, so the client receives the whole
* body with the final DATA frame marked END_STREAM.
*/
public function pacedResumeDeliversWholeBodyTestCase()
{
$site = new H2PacedResumeProbeWebSite("/");
list($key, $connection, $server) =
$this->setUpConnection($site, 131072);
$client = $this->sockets[1];
$stream = 15;
$body = str_repeat("A", 185349);
$site->startPacedBody($key, $server, $connection, $stream,
$body);
$received = $this->drainAndRead($site, $key, $client, "");
list($first_total, ) =
$this->countDataPayload($received, $stream);
$this->assertEqual(131072, $first_total,
'only the first window goes out before any credit returns');
$frame = new WindowUpdateFrame($stream);
$payload = pack("N", 12451840);
$site->handleFrame($key, $server, $connection, $frame,
$payload, 4);
$received = $this->drainAndRead($site, $key, $client, $received);
list($total, $end_seen) =
$this->countDataPayload($received, $stream);
$this->assertEqual(strlen($body), $total,
'the whole body reaches the client after the window update');
$this->assertTrue($end_seen,
'the final data frame carries the end of stream flag');
}
/**
* A response that fits the client's per-stream window is sent in one
* burst, reaching the client whole with the final DATA frame marked
* END_STREAM and needing no WINDOW_UPDATE.
*/
public function oneBurstDeliversWholeBodyTestCase()
{
$site = new H2PacedResumeProbeWebSite("/");
list($key, $connection, $server) =
$this->setUpConnection($site, 131072);
$client = $this->sockets[1];
$stream = 7;
$body = str_repeat("B", 40000);
$site->startPacedBody($key, $server, $connection, $stream,
$body);
$received = $this->drainAndRead($site, $key, $client, "");
list($total, $end_seen) =
$this->countDataPayload($received, $stream);
$this->assertEqual(strlen($body), $total,
'a body within the window is delivered in one burst');
$this->assertTrue($end_seen,
'the final data frame carries the end of stream flag');
}
/**
* The window-update credit is read off the socket through the real
* read path, and the client then closes its send side. Reading the
* credit must resume the parked body, and the end-of-file the close
* raises must not tear the stream down while the rest of the body is
* still queued, so the client receives the whole body with the final
* DATA frame marked END_STREAM. This guards the partial-transfer
* failure where a stalled response, woken only when the socket next
* went readable, was dropped at that wake because end-of-file was
* treated as a reason to close even though bytes were still pending.
*/
public function readCreditThenKeepsFlushingPastEofTestCase()
{
$site = new H2PacedResumeProbeWebSite("/");
list($key, $connection, $server) =
$this->setUpConnection($site, 131072);
$client = $this->sockets[1];
$stream = 15;
$body = str_repeat("A", 185349);
$site->startPacedBody($key, $server, $connection, $stream,
$body);
$received = $this->drainAndRead($site, $key, $client, "");
list($first_total, ) =
$this->countDataPayload($received, $stream);
$this->assertEqual(131072, $first_total,
'only the first window goes out before any credit returns');
$increment = 12451840;
$update = chr(0) . chr(0) . chr(4) . chr(0x08) . chr(0)
. pack("N", $stream & 0x7fffffff) . pack("N", $increment);
fwrite($client, $update);
stream_socket_shutdown($client, STREAM_SHUT_WR);
$site->readOnce($key, $server);
$site->readOnce($key, $server);
$received = $this->drainAndRead($site, $key, $client, $received);
list($total, $end_seen) =
$this->countDataPayload($received, $stream);
$this->assertEqual(strlen($body), $total,
'the whole body still reaches the client after the close');
$this->assertTrue($end_seen,
'the final data frame carries the end of stream flag');
}
/**
* Builds the wire bytes of one HTTP/2 frame: a nine byte header
* (three byte length, one byte type, one byte flags, four byte
* stream id) followed by the payload. The type here is DATA, which
* is enough for a test that only checks the read loop hands each
* framed-off chunk to the frame handler.
*
* @param int $stream the stream id to put in the frame header
* @param string $payload the frame body bytes
* @return string the serialized frame
*/
private function dataFrameBytes($stream, $payload)
{
$length = strlen($payload);
$header = chr(($length >> 16) & 0xff)
. chr(($length >> 8) & 0xff)
. chr($length & 0xff)
. chr(0x00)
. chr(0x00)
. pack("N", $stream & 0x7fffffff);
return $header . $payload;
}
/**
* Two requests that arrive together in a single read must both be
* handed off to the frame handler, not just the first. This guards
* the failure where, after one read carried several streams'
* frames, only the first stream got a response and the rest hung
* because nothing was left to wake the loop and read them.
*/
public function readDispatchesEveryRequestInOneReadTestCase()
{
$site = new H2FrameDrainProbeWebSite("/");
$pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM,
0);
$server = $pair[0];
$client = $pair[1];
stream_set_blocking($server, false);
stream_set_blocking($client, false);
$this->sockets = [$server, $client];
$key = (int)$server;
$connection = new Connection($server, false);
$connection->protocol = 'h2';
$connection->is_websocket = false;
$connection->protocol_state = [];
$site->registerConnection($key, $connection, $server);
$first = $this->dataFrameBytes(1, "abc");
$second = $this->dataFrameBytes(3, "def");
fwrite($client, $first . $second);
$site->readOnce($key, $server);
$this->assertEqual([1, 3], $site->seen,
'both requests arriving in one read are handed off, not '
. 'only the first');
}
/**
* A session written to its file reads back as what was put in it, and
* dropping the session takes the file with it, so an evicted session
* leaves nothing behind on disk.
*/
public function sessionRoundTripsThroughItsFileTestCase()
{
$probe = new SessionFileProbeWebSite();
$probe->useDirectory(sys_get_temp_dir(), 10000);
$identifier = "probe" . getmypid();
$this->assertTrue($probe->write($identifier, ['USER_ID' => 7]),
"a small session is written");
$this->assertTrue(file_exists($probe->pathOf($identifier)),
"its file is there");
$read = $probe->sessionData($identifier);
$this->assertEqual($read['USER_ID'] ?? 0, 7,
"what was put in reads back");
$probe->forget($identifier);
$this->assertTrue(!file_exists($probe->pathOf($identifier)),
"dropping the session takes its file too");
}
/**
* A session grown past what the server accepts as a request body is
* refused rather than written, since holding one that large would cost
* the memory that limit exists to bound.
*/
public function oversizeSessionIsRefusedTestCase()
{
$probe = new SessionFileProbeWebSite();
$probe->useDirectory(sys_get_temp_dir(), 100);
$identifier = "oversize" . getmypid();
$this->assertTrue(!$probe->write($identifier,
['BIG' => str_repeat("x", 500)]),
"a session past the limit is refused");
$this->assertTrue(!file_exists($probe->pathOf($identifier)),
"nothing is written for it");
$this->assertEqual($probe->sessionData($identifier), [],
"and it reads back as empty");
}
/**
* With no period configured, memory sampling stays off and the
* check never reports a sample as due, however much time passes.
*/
public function memorySampleOffWhenUnsetTestCase()
{
$site = new MemorySampleProbeWebSite("/");
$this->assertTrue(!$site->dueAt(1000),
"sampling is off when no period is configured");
$this->assertTrue(!$site->dueAt(500000),
"still off no matter how much time passes");
}
/**
* With a period of zero, which is how an operator turns the feature
* back off, the check still never reports a sample as due.
*/
public function memorySampleOffWhenZeroTestCase()
{
$site = new MemorySampleProbeWebSite("/");
$site->setPeriod(0);
$this->assertTrue(!$site->dueAt(1000),
"a period of zero turns sampling off");
}
/**
* With a positive period the first check is due, further checks are
* not due until the period has passed, and then become due again.
*/
public function memorySampleFiresEveryPeriodTestCase()
{
$site = new MemorySampleProbeWebSite("/");
$site->setPeriod(10);
$this->assertTrue($site->dueAt(1000),
"the first sample is due");
$this->assertTrue(!$site->dueAt(1005),
"a sample is not due before the period elapses");
$this->assertTrue($site->dueAt(1010),
"a sample is due once the period has elapsed");
$this->assertTrue(!$site->dueAt(1011),
"a sample is not due again right away");
}
/**
* Late WINDOW_UPDATE frames for streams the server has already
* finished and closed - which a browser sends as it drains each
* completed response - are ignored rather than stashed as
* pending credit, so a connection that serves a long run of
* responses (a video fetched range by range) never accumulates
* one stranded credit entry per closed stream and never trips
* the pending-credit guard that would close the connection.
*/
public function closedStreamWindowUpdateForClosedStreamIgnoredTestCase()
{
$site = new H2PacedResumeProbeWebSite("/");
list($key, $connection, $server) =
$this->setUpConnection($site, 131072);
$client = $this->sockets[1];
$closed_count = self::CLOSED_STREAM_RUN;
for ($index = 0; $index < $closed_count; $index++) {
$stream = 1 + 2 * $index;
$site->startPacedBody($key, $server, $connection, $stream,
"a small whole body that completes at once");
$this->drainAndRead($site, $key, $client, "");
$frame = new WindowUpdateFrame($stream);
$site->handleFrame($key, $server, $connection, $frame,
pack("N", 65536), 4);
}
$credit = $connection->protocol_state[
'pending_stream_window_credit'] ?? [];
$this->assertEqual(0, count($credit),
'no credit is stashed for streams that already closed');
$this->assertTrue($site->connection($key) !== null,
'the connection is not torn down by a guard tripping');
}
/**
* A WINDOW_UPDATE that arrives for a stream the server has opened
* but not yet begun responding on is still held as pending credit
* and still applied when the response starts, so a browser that
* enlarges a stream's window the instant it opens (the grant that
* lets a larger-than-window page go out in full) is unaffected by
* the closed-stream check.
*/
public function preGrantForOpenStreamStillAppliedTestCase()
{
$site = new H2PacedResumeProbeWebSite("/");
list($key, $connection, $server) =
$this->setUpConnection($site, 65536);
$client = $this->sockets[1];
$stream = 7;
$frame = new WindowUpdateFrame($stream);
$site->handleFrame($key, $server, $connection, $frame,
pack("N", 65536), 4);
$credit = $connection->protocol_state[
'pending_stream_window_credit'] ?? [];
$this->assertEqual(65536, $credit[$stream] ?? 0,
'a pre-grant for an open stream is held, not ignored');
$body = str_repeat("B", 200000);
$site->startPacedBody($key, $server, $connection, $stream,
$body);
$received = $this->drainAndRead($site, $key, $client, "");
list($first_total, ) =
$this->countDataPayload($received, $stream);
$this->assertEqual(131072, $first_total,
'the held pre-grant is added to the initial window so two '
. 'windows worth go out before any further credit');
}
/**
* Gives back the bytes waiting to be written for the test's
* connection.
*
* @return string bytes queued so far
*/
private function queued()
{
$streams = $this->out_streams->getValue($this->site);
return $streams[WebSite::DATA][$this->key] ?? "";
}
/**
* Reads a run of HTTP/2 frames and reports what each one is. Every
* frame is nine bytes of heading and then its body, so a reader can
* walk them without knowing what they carry.
*
* @param string $bytes the queued bytes to read
* @return array one entry per frame, each with its type, flags,
* stream id and body
*/
private function framesIn($bytes)
{
$frames = [];
$offset = 0;
while ($offset + 9 <= strlen($bytes)) {
$length = (ord($bytes[$offset]) << 16) +
(ord($bytes[$offset + 1]) << 8) + ord($bytes[$offset + 2]);
$type = ord($bytes[$offset + 3]);
$flags = ord($bytes[$offset + 4]);
$stream_id = unpack('N',
substr($bytes, $offset + 5, 4))[1] & 0x7FFFFFFF;
$body = substr($bytes, $offset + 9, $length);
$frames[] = ['type' => $type, 'flags' => $flags,
'stream_id' => $stream_id, 'body' => $body];
$offset += 9 + $length;
}
return $frames;
}
/**
* Builds a socket handle that rides the given HTTP/2 stream, or a
* plain one when the stream is zero.
*
* @param int $stream_id the stream the socket rides, zero for none
* @return WebSocket the socket handle
*/
private function socketOnStream($stream_id)
{
$socket = new WebSocket("a-connection", $this->key, $this->site);
$socket->h2_stream_id = $stream_id;
return $socket;
}
/**
* A socket frame small enough to fit goes out as a single data
* frame on its own stream, carrying exactly what was given.
*/
public function smallFrameGoesAsOneDataFrameTestCase()
{
$payload = str_repeat("x", 40);
$this->enqueue_h2->invoke($this->site, $this->key, 5, $payload);
$frames = $this->framesIn($this->queued());
$this->assertEqual(count($frames), 1,
"a small socket frame goes out as one data frame");
$this->assertEqual($frames[0]['type'], 0,
"the frame it goes out in is a data frame");
$this->assertEqual($frames[0]['stream_id'], 5,
"the data frame rides the stream it was given");
$this->assertEqual($frames[0]['body'], $payload,
"the data frame carries what was given, unchanged");
$this->assertEqual($frames[0]['flags'], 0,
"carrying a socket frame does not end the stream");
}
/**
* A socket frame larger than a reader will take in one data frame is
* split across several, and what arrives, joined back up, is what
* was sent. A single oversized frame had been dropped by readers.
*/
public function largeFrameIsSplitToFrameSizeTestCase()
{
$limit = WebSite::H2_MAX_FRAME_SIZE;
$payload = str_repeat("y", 2 * $limit + 100);
$this->enqueue_h2->invoke($this->site, $this->key, 9, $payload);
$frames = $this->framesIn($this->queued());
$this->assertEqual(count($frames), 3,
"a frame over twice the size goes out as three");
$joined = "";
foreach ($frames as $frame) {
$this->assertTrue(strlen($frame['body']) <= $limit,
"no data frame carries more than a reader will take");
$this->assertEqual($frame['stream_id'], 9,
"every piece rides the same stream");
$joined .= $frame['body'];
}
$this->assertEqual($joined, $payload,
"the pieces joined back up are what was sent");
}
/**
* A socket frame of exactly the size a reader will take goes out as
* one data frame, with no empty one following it.
*/
public function frameOfExactlyTheLimitIsNotSplitTestCase()
{
$payload = str_repeat("z", WebSite::H2_MAX_FRAME_SIZE);
$this->enqueue_h2->invoke($this->site, $this->key, 3, $payload);
$frames = $this->framesIn($this->queued());
$this->assertEqual(count($frames), 1,
"a frame of exactly the limit goes out as one");
$this->assertEqual(strlen($frames[0]['body']),
WebSite::H2_MAX_FRAME_SIZE,
"that one frame carries the whole of it");
}
/**
* An answer sent on a socket that rides an HTTP/2 stream is wrapped
* in a data frame. Writing it as raw bytes had left those bytes
* where a frame heading was looked for, so everything sent after it
* on that connection could not be read.
*/
public function answerOnACarriedSocketIsWrappedTestCase()
{
$socket = $this->socketOnStream(11);
$this->send_frame->invoke($this->site, $this->key, $socket,
"\x88\x02\x03\xe8");
$frames = $this->framesIn($this->queued());
$this->assertEqual(count($frames), 1,
"the answer goes out as one data frame");
$this->assertEqual($frames[0]['type'], 0,
"the answer is wrapped in a data frame");
$this->assertEqual($frames[0]['stream_id'], 11,
"the wrapping rides the socket's own stream");
$this->assertEqual($frames[0]['body'], "\x88\x02\x03\xe8",
"the wrapping carries the answer unchanged");
}
/**
* An answer sent on a socket that has a connection to itself goes
* out as it stands, since there is no framing around it to keep.
*/
public function answerOnAPlainSocketGoesAsItStandsTestCase()
{
$socket = $this->socketOnStream(0);
$this->send_frame->invoke($this->site, $this->key, $socket,
"\x8a\x00");
$this->assertEqual($this->queued(), "\x8a\x00",
"a plain socket's answer goes out as it stands");
}
/**
* A write for a connection that has been taken down is let go
* rather than reaching for a connection that is no longer there.
*/
public function writeForAGoneConnectionIsDroppedTestCase()
{
$this->in_streams->setValue($this->site,
[WebSite::CONNECTION => []]);
$socket = $this->socketOnStream(13);
$this->send_frame->invoke($this->site, $this->key, $socket,
"\x81\x01a");
$this->assertEqual($this->queued(), "",
"nothing is queued for a connection that is gone");
}
/**
* Builds a connection standing at a given protocol, saying whether
* it carries a socket.
*
* @param string $protocol what the connection speaks
* @param bool $is_websocket whether it carries a socket
* @return Connection the connection
*/
private function connectionSpeaking($protocol, $is_websocket)
{
$reflection = new \ReflectionClass(Connection::class);
$connection = $reflection->newInstanceWithoutConstructor();
$connection->protocol = $protocol;
$connection->is_websocket = $is_websocket;
return $connection;
}
/**
* Checks that a reply named as a file is drained from disk a piece at
* a time rather than read into a string. A pack built for a git pull
* can be larger than the memory the server has, and gathering it as a
* string exhausted the process on a live site.
*/
public function fileBackedReplyDrainsInPiecesTestCase()
{
$site = new WebSite();
$path = tempnam(sys_get_temp_dir(), "yioop_reply_");
$piece = str_repeat("a line of the reply\n", 20000);
$handle = fopen($path, "wb");
for ($count = 0; $count < 4; $count++) {
fwrite($handle, $piece);
}
fclose($handle);
$whole = filesize($path);
$key = "test-file-reply";
$naming = new \ReflectionMethod($site, "sendRestFromFile");
$this->assertTrue($naming->invoke($site, $key, $path),
"a reply may be named as a file");
$refill = new \ReflectionMethod($site, "refillFromFile");
$reading = new \ReflectionProperty($site, "out_streams");
$sent = 0;
$largest = 0;
for ($step = 0; $step < 200; $step++) {
$refill->invoke($site, $key);
$streams = $reading->getValue($site);
$held = $streams[WebSite::DATA][$key] ?? "";
if (empty($streams[WebSite::DATA_FILE][$key]) && $held === "") {
break;
}
$sent += strlen($held);
$largest = max($largest, strlen($held));
$streams[WebSite::DATA][$key] = "";
$reading->setValue($site, $streams);
}
unlink($path);
$this->assertEqual($sent, $whole,
"every byte of the file goes out");
$this->assertTrue($largest <= WebSite::OUT_FILE_STEP,
"no more than one piece is held at a time");
}
/**
* A connection speaking HTTP/2 keeps to the HTTP/2 way of finishing
* a write even while it carries a socket, since the rest of a large
* reply is still to be sent on it. A connection upgraded from
* HTTP/1.1 carries its socket alone and takes the socket way.
*/
public function drainPathFollowsTheProtocolTestCase()
{
$this->assertTrue(!$this->uses_socket_drain->invoke($this->site,
$this->connectionSpeaking('h2', true)),
"a connection speaking HTTP/2 keeps to the HTTP/2 way");
$this->assertTrue($this->uses_socket_drain->invoke($this->site,
$this->connectionSpeaking('ws', true)),
"a connection upgraded to a socket takes the socket way");
$this->assertTrue(!$this->uses_socket_drain->invoke($this->site,
$this->connectionSpeaking('h1', false)),
"an ordinary connection takes neither");
$this->assertTrue(!$this->uses_socket_drain->invoke($this->site,
null),
"a connection already taken down takes neither");
}
}
(c) 2026 Seekquarry - Open Source Search Engine Software