logdash-io · mleczakm · Aug 8, 2026
diff --git a/src/Metrics/Metrics.php b/src/Metrics/Metrics.php
@@ -5,18 +5,43 @@
 namespace Logdash\Metrics;
 
 use GuzzleHttp\Client;
-use GuzzleHttp\Exception\GuzzleException;
+use GuzzleHttp\Pool;
+use GuzzleHttp\Promise\PromiseInterface;
 use Logdash\Logger\InternalLogger;
+use Logdash\Queue\RequestQueue;
 use Logdash\Types\RequiredInitializationParams;
+use Throwable;
 
 class Metrics implements BaseMetrics
 {
+    private const BATCH_SIZE = 20;
+    private const FLUSH_INTERVAL_MS = 1000;
+    private const CONCURRENCY = 10;
+
     private Client $httpClient;
+    private RequestQueue $queue;
 
     public function __construct(
-        private readonly RequiredInitializationParams $params
+        private readonly RequiredInitializationParams $params,
+        ?Client $httpClient = null
     ) {
-        $this->httpClient = new Client();
+        $this->httpClient = $httpClient ?? new Client([
+            'connect_timeout' => 2,
+            'timeout' => 5,
+        ]);
+
+        $this->queue = new RequestQueue(
+            sendFn: function (array $metrics): void {
+                $this->sendBatch($metrics);
+            },
+            batchSize: self::BATCH_SIZE,
+            flushIntervalMs: self::FLUSH_INTERVAL_MS,
+            // Metric updates (especially `mutate`) aren't idempotent, so a
+            // failed item is never retried here - only dispatched
+            // concurrently instead of one-by-one.
+            maxAttempts: 1,
+            verbose: $this->params->verbose,
+        );
     }
 
     public function set(string $name, float $value): void
@@ -25,7 +50,7 @@ public function set(string $name, float $value): void
             InternalLogger::getInternalLogger()->verbose("Setting metric {$name} to {$value}");
         }
 
-        $this->sendMetric($name, $value, MetricOperation::SET);
+        $this->enqueue($name, $value, MetricOperation::SET);
     }
 
     public function mutate(string $name, float $value): void
@@ -34,29 +59,55 @@ public function mutate(string $name, float $value): void
             InternalLogger::getInternalLogger()->verbose("Mutating metric {$name} by {$value}");
         }
 
-        $this->sendMetric($name, $value, MetricOperation::CHANGE);
+        $this->enqueue($name, $value, MetricOperation::CHANGE);
+    }
+
+    /**
+     * Sends any buffered metric updates immediately instead of waiting for
+     * the batch size/interval or process shutdown. Useful before a
+     * long-running worker goes idle.
+     */
+    public function flush(): void
+    {
+        $this->queue->flush();
     }
 
-    private function sendMetric(string $name, float $value, MetricOperation $operation): void
+    private function enqueue(string $name, float $value, MetricOperation $operation): void
     {
-        try {
-            $this->httpClient->put($this->params->host . '/metrics', [
-                'headers' => [
-                    'Content-Type' => 'application/json',
-                    'project-api-key' => $this->params->apiKey,
-                ],
-                'json' => [
-                    'name' => $name,
-                    'value' => $value,
-                    'operation' => $operation->value,
-                ],
-                'timeout' => 5,
-            ]);
-        } catch (GuzzleException $e) {
-            if ($this->params->verbose) {
-                InternalLogger::getInternalLogger()->verbose("Failed to send metric: " . $e->getMessage());
+        $this->queue->add([
+            'name' => $name,
+            'value' => $value,
+            'operation' => $operation->value,
+        ]);
+    }
+
+    /**
+     * @param list<array{name: string, value: float, operation: string}> $metrics
+     */
+    private function sendBatch(array $metrics): void
+    {
+        $requests = (function () use ($metrics) {
+            foreach ($metrics as $metric) {
+                yield fn (): PromiseInterface => $this->httpClient->putAsync($this->params->host . '/metrics', [
+                    'headers' => [
+                        'Content-Type' => 'application/json',
+                        'project-api-key' => $this->params->apiKey,
+                    ],
+                    'json' => $metric,
+                ]);
             }
-            // Fail silently in production
-        }
+        })();
+
+        $pool = new Pool($this->httpClient, $requests, [
+            'concurrency' => self::CONCURRENCY,
+            'rejected' => function (Throwable $reason): void {
+                if ($this->params->verbose) {
+                    InternalLogger::getInternalLogger()->verbose('Failed to send metric: ' . $reason->getMessage());
+                }
+                // Fail silently in production
+            },
+        ]);
+
+        $pool->promise()->wait();
     }
 }
diff --git a/src/Queue/RequestQueue.php b/src/Queue/RequestQueue.php
@@ -0,0 +1,101 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Logdash\Queue;
+
+use Logdash\Logger\InternalLogger;
+use Throwable;
+
+/**
+ * Buffers items and flushes them in batches instead of sending one HTTP
+ * request per item. A batch is sent once it reaches $batchSize, once
+ * $flushIntervalMs has elapsed since the last flush, or when the PHP
+ * process/request ends (via a registered shutdown flush).
+ */
+final class RequestQueue
+{
+    /** @var list<mixed> */
+    private array $items = [];
+
+    private bool $shutdownRegistered = false;
+
+    private float $lastFlushAt;
+
+    /**
+     * @param callable(list<mixed>): void $sendFn
+     */
+    public function __construct(
+        private readonly mixed $sendFn,
+        private readonly int $batchSize = 25,
+        private readonly int $flushIntervalMs = 2000,
+        private readonly int $maxAttempts = 1,
+        private readonly int $retryDelayMs = 200,
+        private readonly bool $verbose = false,
+    ) {
+        $this->lastFlushAt = microtime(true);
+    }
+
+    public function add(mixed $item): void
+    {
+        $this->items[] = $item;
+        $this->registerShutdownFlush();
+
+        $elapsedMs = (microtime(true) - $this->lastFlushAt) * 1000;
+
+        if (count($this->items) >= $this->batchSize || $elapsedMs >= $this->flushIntervalMs) {
+            $this->flush();
+        }
+    }
+
+    public function flush(): void
+    {
+        $this->lastFlushAt = microtime(true);
+
+        if ($this->items === []) {
+            return;
+        }
+
+        $batch = $this->items;
+        $this->items = [];
+
+        $this->sendWithRetry($batch);
+    }
+
+    /**
+     * @param list<mixed> $batch
+     */
+    private function sendWithRetry(array $batch): void
+    {
+        for ($attempt = 1; $attempt <= $this->maxAttempts; $attempt++) {
+            try {
+                ($this->sendFn)($batch);
+                return;
+            } catch (Throwable $e) {
+                if ($attempt >= $this->maxAttempts) {
+                    if ($this->verbose) {
+                        InternalLogger::getInternalLogger()->verbose(
+                            'Failed to flush batch of ' . count($batch) . " item(s) after {$attempt} attempt(s): "
+                            . $e->getMessage()
+                        );
+                    }
+                    return;
+                }
+
+                usleep($this->retryDelayMs * 1000);
+            }
+        }
+    }
+
+    private function registerShutdownFlush(): void
+    {
+        if ($this->shutdownRegistered) {
+            return;
+        }
+
+        $this->shutdownRegistered = true;
+        register_shutdown_function(function (): void {
+            $this->flush();
+        });
+    }
+}
diff --git a/src/Sync/HttpLogSync.php b/src/Sync/HttpLogSync.php
@@ -5,49 +5,72 @@
 namespace Logdash\Sync;
 
 use GuzzleHttp\Client;
-use GuzzleHttp\Exception\GuzzleException;
-use Logdash\Logger\InternalLogger;
+use Logdash\Queue\RequestQueue;
 use Logdash\Types\LogLevel;
 use Logdash\Types\RequiredInitializationParams;
 
 class HttpLogSync implements LogSync
 {
+    private const BATCH_SIZE = 25;
+    private const FLUSH_INTERVAL_MS = 2000;
+
     private int $sequenceNumber = 0;
     private Client $httpClient;
+    private RequestQueue $queue;
 
     public function __construct(
-        private readonly RequiredInitializationParams $params
+        private readonly RequiredInitializationParams $params,
+        ?Client $httpClient = null
     ) {
-        $this->httpClient = new Client();
+        $this->httpClient = $httpClient ?? new Client([
+            'connect_timeout' => 2,
+            'timeout' => 5,
+        ]);
+
+        $this->queue = new RequestQueue(
+            sendFn: function (array $logs): void {
+                $this->sendBatch($logs);
+            },
+            batchSize: self::BATCH_SIZE,
+            flushIntervalMs: self::FLUSH_INTERVAL_MS,
+            maxAttempts: 2,
+            verbose: $this->params->verbose,
+        );
+    }
+
+    public function send(string $message, LogLevel $level, string $createdAt): void
+    {
+        $this->queue->add([
+            'message' => $message,
+            'level' => $level->value,
+            'createdAt' => $createdAt,
+            'sequenceNumber' => $this->sequenceNumber++,
+        ]);
     }
 
     /**
-     * TODO:
-     * - queue
-     * - retry
-     * - batching
+     * Sends any buffered logs immediately instead of waiting for the batch
+     * size/interval or process shutdown. Useful before a long-running
+     * worker goes idle.
      */
-    public function send(string $message, LogLevel $level, string $createdAt): void
+    public function flush(): void
+    {
+        $this->queue->flush();
+    }
+
+    /**
+     * @param list<array{message: string, level: string, createdAt: string, sequenceNumber: int}> $logs
+     */
+    private function sendBatch(array $logs): void
     {
-        try {
-            $this->httpClient->post($this->params->host . '/logs', [
-                'headers' => [
-                    'Content-Type' => 'application/json',
-                    'project-api-key' => $this->params->apiKey,
-                ],
-                'json' => [
-                    'message' => $message,
-                    'level' => $level->value,
-                    'createdAt' => $createdAt,
-                    'sequenceNumber' => $this->sequenceNumber++,
-                ],
-                'timeout' => 5,
-            ]);
-        } catch (GuzzleException $e) {
-            if ($this->params->verbose) {
-                InternalLogger::getInternalLogger()->verbose("Failed to send log: " . $e->getMessage());
-            }
-            // Fail silently in production
-        }
+        $this->httpClient->post($this->params->host . '/logs/batch', [
+            'headers' => [
+                'Content-Type' => 'application/json',
+                'project-api-key' => $this->params->apiKey,
+            ],
+            'json' => [
+                'logs' => $logs,
+            ],
+        ]);
     }
 }