Add rate-limiter to forwarder

bin/sentry-agent

@@ -63,7 +63,13 @@ $queue = new EnvelopeQueue(
     $upstreamConcurrency,
     $queueLimit,
     function (Envelope $envelope) use ($forwarder) {
-        return $forwarder->forward($envelope);
+        try {
+            return $forwarder->forward($envelope);
+        } catch (Exception $e) {
+            Log::error("Failed to forward envelope: {$e->getMessage()}");
+
+            return new React\Promise\Internal\RejectedPromise($e);
+        }
     }
 );
 

composer.json

@@ -12,6 +12,7 @@
     ],
     "require": {
         "php": "^7.2|^8",
+        "ext-json": "*",
         "clue/mq-react": "^1.6",
         "react/http": "^1.11",
         "react/socket": "^1.16",
@@ -24,16 +25,24 @@
     },
     "require-dev": {
         "friendsofphp/php-cs-fixer": "^3.70",
-        "phpstan/phpstan": "^2.1"
+        "phpstan/phpstan": "^2.1",
+        "phpunit/phpunit": "^8.5|^9.6"
+    },
+    "autoload-dev": {
+        "psr-4": {
+            "Sentry\\Agent\\Tests\\": "tests/"
+        }
     },
     "bin": [
         "bin/sentry-agent"
     ],
     "scripts": {
         "check": [
             "@cs-check",
-            "@phpstan"
+            "@phpstan",
+            "@tests"
         ],
+        "tests": "vendor/bin/phpunit --verbose",
         "cs-check": "vendor/bin/php-cs-fixer fix --verbose --diff --dry-run",
         "cs-fix": "vendor/bin/php-cs-fixer fix --verbose --diff",
         "phpstan": "vendor/bin/phpstan analyse"

phpunit.xml

@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<phpunit xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:noNamespaceSchemaLocation="vendor/phpunit/phpunit/phpunit.xsd"
+         backupGlobals="true"
+         colors="true"
+         bootstrap="vendor/autoload.php"
+         cacheResult="false"
+         beStrictAboutOutputDuringTests="true"
+>
+    <testsuites>
+        <testsuite name="unit">
+            <directory>tests</directory>
+        </testsuite>
+    </testsuites>
+
+    <coverage>
+        <include>
+            <directory suffix=".php">src</directory>
+        </include>
+    </coverage>
+</phpunit>

src/Envelope.php

@@ -4,51 +4,182 @@
 
 namespace Sentry\Agent;
 
+use Sentry\Agent\Exceptions\MalformedEnvelope;
+use Sentry\Dsn;
+
 /**
  * @internal
+ *
+ * @phpstan-type EnvelopeHeader array{
+ *     dsn: string,
+ * }
  */
 class Envelope
 {
+    public const CONTENT_TYPE = 'application/x-sentry-envelope';
+
+    /**
+     * @var EnvelopeHeader The envelope header
+     */
+    private $header;
+
     /**
-     * @var string
+     * @var EnvelopeItem[] The envelope items
      */
-    private $data;
+    private $items;
 
-    public function __construct(string $data)
+    /**
+     * @param EnvelopeHeader $header
+     * @param EnvelopeItem[] $items
+     */
+    public function __construct(array $header, array $items)
     {
-        $this->data = $data;
+        $this->header = $header;
+        $this->items = $items;
     }
 
-    public function getDsn(): ?string
+    /**
+     * @return EnvelopeHeader
+     */
+    public function getHeader(): array
     {
-        $header = $this->getHeader();
-
-        if ($header === null) {
-            return null;
-        }
+        return $this->header;
+    }
 
-        $parsedHeader = json_decode($header, true);
+    public function getDsn(): Dsn
+    {
+        return Dsn::createFromString($this->header['dsn']);
+    }
 
-        if (\is_array($parsedHeader) && !empty($parsedHeader['dsn']) && \is_string($parsedHeader['dsn'])) {
-            return $parsedHeader['dsn'];
-        }
+    /**
+     * @return EnvelopeItem[]
+     */
+    public function getItems(): array
+    {
+        return $this->items;
+    }
 
-        return null;
+    /**
+     * @param callable(EnvelopeItem): bool $callback if the callback returns true, the item will be removed from the envelope
+     */
+    public function rejectItems(callable $callback): void
+    {
+        $this->items = array_filter(
+            $this->items,
+            static function (EnvelopeItem $item) use ($callback) {
+                return !$callback($item);
+            }
+        );
     }
 
-    public function getData(): string
+    public function __toString()
     {
-        return $this->data;
+        $data = implode(
+            "\n",
+            array_map(
+                static function (EnvelopeItem $item): string {
+                    return (string) $item;
+                }, $this->items
+            )
+        );
+
+        // We always terminate with an additional newline
+        return json_encode($this->header) . "\n{$data}";
     }
 
-    public function getHeader(): ?string
+    /**
+     * @throws MalformedEnvelope
+     */
+    public static function fromString(string $envelope): self
     {
-        $position = strpos($this->data, "\n");
+        $consumePart = static function () use (&$envelope): ?string {
+            // Once we fully consumed the envelope, we return null indicating EOF
+            if ($envelope === '') {
+                return null;
+            }
+
+            // Parts are newline delimited so we can find the next newline to find the end of the next part
+            $nextNewline = strpos($envelope, "\n");
+
+            if ($nextNewline === false) {
+                $nextNewline = \strlen($envelope);
+            }
+
+            $part = substr($envelope, 0, $nextNewline);
+
+            // We consume the newline as well
+            $envelope = substr($envelope, $nextNewline + 1);
+
+            // Empty parts are additional trailing newlines, which can be ignored
+            if ($part === '') {
+                return null;
+            }
+
+            return $part;
+        };
+
+        $consumeBytes = static function (int $bytes) use (&$envelope): string {
+            if (\strlen($envelope) < $bytes) {
+                throw new MalformedEnvelope('Envelope reached EOF before consuming expected bytes');
+            }
+
+            $part = substr($envelope, 0, $bytes);
+
+            $envelope = substr($envelope, $bytes + 1);
+
+            return $part;
+        };
+
+        $parseJson = static function (?string $json): array {
+            if ($json === null) {
+                throw new MalformedEnvelope('Envelope reached EOF before consuming expected JSON');
+            }
+
+            $decoded = json_decode($json, true);
+
+            if (!\is_array($decoded)) {
+                // Technically we could have a non-JSON error here (if we try to parse a single JSON scalar for example)
+                // but we don't really care if that happens and we can just assume there was a problem parsing the JSON if we don't get an array
+                throw new MalformedEnvelope('Failed to decode JSON: ' . json_last_error_msg());
+            }
+
+            return $decoded;
+        };
+
+        // The first part is always the envelope header
+        $header = $parseJson($consumePart());
+
+        // Technically the header could not contain the DSN key, but we don't really care about that case since we won't be able to forward the envelope
+        if (!isset($header['dsn'])) {
+            throw new MalformedEnvelope('Envelope header does not contain a DSN');
+        }
+
+        $items = [];
+
+        while ($rawItemHeader = $consumePart()) {
+            $itemHeader = $parseJson($rawItemHeader);
+
+            // The item header should always contain the type
+            if (!isset($itemHeader['type'])) {
+                throw new MalformedEnvelope('Envelope item header does not contain a type');
+            }
+
+            // The size in the header is optional
+            $itemContentLength = $itemHeader['length'] ?? null;
+
+            if ($itemContentLength === null) {
+                $itemContent = $consumePart();
+
+                if ($itemContent === null) {
+                    throw new MalformedEnvelope('Envelope reached EOF before consuming expected item content');
+                }
+            } else {
+                $itemContent = $consumeBytes($itemContentLength);
+            }
 
-        if ($position === false) {
-            return null;
+            $items[] = new EnvelopeItem($itemHeader, $itemContent);
         }
 
-        return substr($this->data, 0, $position);
+        return new self($header, $items);
     }
 }

src/EnvelopeForwarder.php

@@ -8,6 +8,8 @@
 use React\Http\Browser;
 use React\Promise\PromiseInterface;
 use Sentry\Dsn;
+use Sentry\HttpClient\Response;
+use Sentry\Transport\RateLimiter;
 
 /**
  * @internal
@@ -44,6 +46,11 @@ class EnvelopeForwarder
      */
     private $onEnvelopeError;
 
+    /**
+     * @var array<string, RateLimiter>
+     */
+    private $rateLimiters = [];
+
     /**
      * @param callable(ResponseInterface): void $onEnvelopeSent  called when the envelope is sent
      * @param callable(\Throwable): void        $onEnvelopeError called when the envelope fails to send
@@ -62,11 +69,20 @@ public function forward(Envelope $envelope): PromiseInterface
     {
         $dsn = $envelope->getDsn();
 
-        if ($dsn === null) {
-            throw new \RuntimeException('The envelope does not contain a DSN.');
-        }
+        $rateLimiter = $this->getRateLimiter($dsn);
+
+        $envelope->rejectItems(static function (EnvelopeItem $envelopeItem) use ($rateLimiter) {
+            $envelopeItemType = $envelopeItem->getItemType();
+
+            // @TODO: We should make the rate limiter accept an arbitrary item type to allow for flexibility when adding new item types
+            if ($envelopeItemType === null) {
+                return false;
+            }
+
+            return $rateLimiter->isRateLimited($envelopeItemType);
+        });
 
-        $dsn = Dsn::createFromString($dsn);
+        // @TODO: If we rate limit all the items we have an empty envelope which we should not send and just return
 
         $authHeader = [
             'sentry_version=' . self::PROTOCOL_VERSION,
@@ -76,10 +92,32 @@ public function forward(Envelope $envelope): PromiseInterface
 
         // @TODO: Implement any number of missing options like the user-agent, encoding, proxy etc.
 
-        return (new Browser())->withTimeout($this->timeout)->post($dsn->getEnvelopeApiEndpointUrl(), [
-            'User-Agent' => self::IDENTIFIER . '/' . self::VERSION,
-            'Content-Type' => 'application/x-sentry-envelope',
-            'X-Sentry-Auth' => 'Sentry ' . implode(', ', $authHeader),
-        ], $envelope->getData())->then($this->onEnvelopeSent, $this->onEnvelopeError);
+        // @TODO: We might want to replace this Browser API with a cURL implementation using curl_multi_exec
+        return (new Browser())->withTimeout($this->timeout)->post(
+            $dsn->getEnvelopeApiEndpointUrl(),
+            [
+                'User-Agent' => self::IDENTIFIER . '/' . self::VERSION,
+                'Content-Type' => Envelope::CONTENT_TYPE,
+                'X-Sentry-Auth' => 'Sentry ' . implode(', ', $authHeader),
+            ],
+            (string) $envelope
+        )->then(function (ResponseInterface $response) use ($rateLimiter) {
+            $rateLimiter->handleResponse(
+                new Response($response->getStatusCode(), $response->getHeaders(), $response->getStatusCode() > 400 ? $response->getBody()->getContents() : '')
+            );
+
+            \call_user_func($this->onEnvelopeSent, $response);
+        }, $this->onEnvelopeError);
+    }
+
+    private function getRateLimiter(Dsn $dsn): RateLimiter
+    {
+        $key = $dsn->getEnvelopeApiEndpointUrl();
+
+        if (!isset($this->rateLimiters[$key])) {
+            $this->rateLimiters[$key] = new RateLimiter();
+        }
+
+        return $this->rateLimiters[$key];
     }
 }