Add Priority handling for HTTP/3

amphp · Jan 13, 2024 · a3dbb24 · a3dbb24
1 parent eb70ad6
commit a3dbb24
Show file tree

Hide file tree

Showing 15 changed files with 858 additions and 6 deletions.
diff --git a/composer.json b/composer.json
@@ -50,6 +50,7 @@
         "amphp/http-client": "^5",
         "amphp/log": "^2",
         "amphp/php-cs-fixer-config": "^2",
+        "httpwg/structured-field-tests": "1.0",
         "league/uri-components": "^2.4.2 | ^7.1",
         "monolog/monolog": "^3",
         "phpunit/phpunit": "^9",
@@ -82,6 +83,20 @@
         "test": "@php -dzend.assertions=1 -dassert.exception=1 ./vendor/bin/phpunit",
         "code-style": "@php ./vendor/bin/php-cs-fixer fix"
     },
+    "repositories": [
+        {
+            "type": "package",
+            "package": {
+                "name": "httpwg/structured-field-tests",
+                "version": "1.0",
+                "source": {
+                    "url": "https://github.com/httpwg/structured-field-tests",
+                    "type": "git",
+                    "reference": "origin/main"
+                }
+            }
+        }
+    ],
     "config": {
         "allow-plugins": {
             "ocramius/package-versions": false

diff --git a/src/Driver/Http3Driver.php b/src/Driver/Http3Driver.php
@@ -2,7 +2,6 @@
 
 namespace Amp\Http\Server\Driver;
 
-use Amp\ByteStream\ClosedException;
 use Amp\ByteStream\ReadableIterableStream;
 use Amp\CancelledException;
 use Amp\DeferredCancellation;
@@ -255,7 +254,7 @@ public function handleConnection(Client $client, QuicConnection|Socket $connecti
                                 }
 
                                 $trailerDeferred = new DeferredFuture;
-                                $bodyQueue = new Queue();
+                                $bodyQueue = new Queue;
 
                                 try {
                                     $trailers = new Trailers(
@@ -296,6 +295,10 @@ public function handleConnection(Client $client, QuicConnection|Socket $connecti
                                     $expectedLength = null;
                                 }
 
+                                if (isset($headers["priority"])) {
+                                    $this->updatePriority($stream, $headers["priority"]);
+                                }
+
                                 $dataSuspension = null;
                                 $body = new RequestBody(
                                     new ReadableIterableStream($bodyQueue->pipe()),
@@ -439,6 +442,18 @@ function (int $bodySize) use (&$bodySizeLimit, &$dataSuspension) {
                         $parser->abort(new Http3ConnectionException("A push stream must not be initiated by the client", Http3Error::H3_STREAM_CREATION_ERROR));
                         break;
 
+                    case Http3Frame::PRIORITY_UPDATE_Request:
+                        [, $streamId, $structuredUpdate] = $frame;
+                        // The RFC says we _should_ temporarily buffer unknown stream ids. We currently don't for simplicity. To eventually improve?
+                        if ($stream = $connection->getStream($streamId)) {
+                            $this->updatePriority($stream, $structuredUpdate);
+                        }
+                        break;
+
+                    case Http3Frame::PRIORITY_UPDATE_Push:
+                        $parser->abort(new Http3ConnectionException("No PRIORITY_UPDATE frame may be sent for unpromised push streams", Http3Error::H3_ID_ERROR));
+                        break;
+
                     default:
                         $parser->abort(new Http3ConnectionException("An unexpected stream or frame was received", Http3Error::H3_FRAME_UNEXPECTED));
                 }
@@ -451,6 +466,12 @@ function (int $bodySize) use (&$bodySizeLimit, &$dataSuspension) {
         }
     }
 
+    public function updatePriority(QuicSocket $socket, array|string $headers)
+    {
+        [$urgency, $incremental] = Http3Parser::parsePriority($headers);
+        $socket->setPriority($urgency + 124 /* 127 is default for QUIC, 3 is default for HTTP */, $incremental);
+    }
+
     public function getPendingRequestCount(): int
     {
         return $this->requestStreams->count();
@@ -473,7 +494,6 @@ public function stop(): void
             $this->highestStreamId,
         )) || true);
 
-
         $outstanding = $this->requestStreams->count();
         if ($outstanding === 0) {
             $this->writer->close();

diff --git a/src/Driver/Internal/Http3/Http3Frame.php b/src/Driver/Internal/Http3/Http3Frame.php
@@ -13,5 +13,5 @@ enum Http3Frame: int
     case ORIGIN = 0x0c;
     case MAX_PUSH_ID = 0x0d;
     case PRIORITY_UPDATE_Request = 0xF0700;
-    case PRIORITY_UPDATE_Response = 0xF0701;
+    case PRIORITY_UPDATE_Push = 0xF0701;
 }
diff --git a/src/Driver/Internal/Http3/Http3Parser.php b/src/Driver/Internal/Http3/Http3Parser.php
@@ -3,6 +3,8 @@
 namespace Amp\Http\Server\Driver\Internal\Http3;
 
 use Amp\Http\Http2\Http2Parser;
+use Amp\Http\Server\Driver\Internal\Http3\Rfc8941\Boolean;
+use Amp\Http\Server\Driver\Internal\Http3\Rfc8941\Number;
 use Amp\Pipeline\ConcurrentIterator;
 use Amp\Pipeline\Queue;
 use Amp\Quic\QuicConnection;
@@ -310,7 +312,7 @@ public function process(): ConcurrentIterator
                                             return;
                                         }
 
-                                        if ($frame !== Http3Frame::GOAWAY || $frame !== Http3Frame::MAX_PUSH_ID || $frame !== Http3Frame::CANCEL_PUSH) {
+                                        if ($frame !== Http3Frame::GOAWAY && $frame !== Http3Frame::MAX_PUSH_ID && $frame !== Http3Frame::CANCEL_PUSH && $frame !== Http3Frame::PRIORITY_UPDATE_Request && $frame !== Http3Frame::PRIORITY_UPDATE_Push) {
                                             throw new Http3ConnectionException("An unexpected frame was received on the control stream", Http3Error::H3_FRAME_UNEXPECTED);
                                         }
 
@@ -321,7 +323,11 @@ public function process(): ConcurrentIterator
                                             }
                                             return;
                                         }
-                                        $this->queue->push([$frame, $id]);
+                                        if ($frame === Http3Frame::PRIORITY_UPDATE_Request || $frame === Http3Frame::PRIORITY_UPDATE_Push) {
+                                            $this->queue->push([$frame, $id, \substr($contents, $tmpOff)]);
+                                        } else {
+                                            $this->queue->push([$frame, $id]);
+                                        }
                                     }
 
                                     // no break
@@ -368,6 +374,31 @@ public function process(): ConcurrentIterator
         return $this->queue->iterate();
     }
 
+    // Note: format is shared with HTTP/2
+    public static function parsePriority(array|string $headers): ?array
+    {
+        $urgency = 3;
+        $incremental = false;
+        if ($priority = Rfc8941::parseDictionary($headers)) {
+            if (isset($priority["u"])) {
+                $number = $priority["u"];
+                if ($number instanceof Number) {
+                    $value = $number->item;
+                    if (\is_int($value) && $value >= 0 && $value <= 7) {
+                        $urgency = $number->item;
+                    }
+                }
+            }
+            if (isset($priority["i"])) {
+                $bool = $priority["i"];
+                if ($bool instanceof Boolean) {
+                    $incremental = $bool->item;
+                }
+            }
+        }
+        return [$urgency, $incremental];
+    }
+
     public function abort(Http3ConnectionException $exception)
     {
         if (!$this->queue->isComplete()) {