ale94lko / php-queue-prophet
Lightweight, framework-agnostic PHP library to predict background worker memory leaks and queue overflow time before process crashes occur.
Requires
- php: ^8.1
Requires (Dev)
- friendsofphp/php-cs-fixer: ^3.64
- phpstan/phpstan: ^2.1
- phpunit/phpunit: ^10.5
README
Lightweight, zero-dependency, framework-agnostic PHP library that predicts background worker memory leaks (OOM) and queue Time-to-Overflow (TTO) using ordinary least-squares linear regression over a fixed-size sliding window.
Requirements
- PHP 8.1+
Installation
composer require ale94lko/php-queue-prophet
Try it locally (playground)
cd examples/playground
composer install
composer demo
See examples/playground/README.md for memory-leak, stable-memory, and queue TTO demos.
Features
- Memory leak forecasting — estimates remaining job iterations before a worker hits its memory limit
- Queue Time-to-Overflow (TTO) — projects when a backlog will breach capacity given arrival vs processing rates
- Framework agnostic — Vanilla PHP, Laravel Queue, Symfony Messenger, RoadRunner, Swoole, etc.
- Zero runtime dependencies — pure PHP math; the predictor itself cannot leak (bounded sliding window)
- Strict typing — PHPStan level 8,
declare(strict_types=1)everywhere
Quick start — worker memory health
use PhpQueueProphet\WorkerHealthPredictor; $predictor = new WorkerHealthPredictor( memoryLimit: '128M', // or an int in bytes sampleWindowSize: 20, ); while ($job = $queue->pop()) { $job->handle(); $predictor->recordSample(); $remaining = $predictor->predictRemainingJobs(); // null → not enough samples yet (< 3) // INF → no leak detected (flat or shrinking memory) if ($remaining !== null && $remaining < 50) { // Gracefully stop / recycle the worker break; } }
Health report
$report = $predictor->generateHealthReport(); $report->leakRatePerJobBytes; // slope m (bytes per job) $report->estimatedRemainingJobs; $report->rSquared; // fit quality [0, 1] $report->isAtRisk(50); // remaining jobs < threshold? $report->getMemoryUsagePercent();
Injecting samples (tests / custom collectors)
$predictor->recordSample(memory_get_usage(true)); // or a custom value: $predictor->recordSample(12_582_912);
Quick start — queue Time-to-Overflow
use PhpQueueProphet\QueueOverflowPredictor; $tracker = new QueueOverflowPredictor(maxCapacity: 10_000); // Rates in items/second (from your broker metrics, Redis LLEN deltas, etc.) $tracker->record( currentDepth: 4_200, arrivalRate: 80.0, processingRate: 50.0, ); $seconds = $tracker->predictTimeToOverflow(); // → 193.333… ((10000 - 4200) / 30) if ($tracker->getMetrics()->isAtRisk(300)) { // Scale workers, alert, shed load… }
Framework integration
The library has no framework bindings. Hook it where your workers already run.
Laravel Queue
Record a sample after each job finishes (e.g. in your worker process or a listener):
use Illuminate\Queue\Events\JobProcessed; use Illuminate\Support\Facades\Event; use PhpQueueProphet\WorkerHealthPredictor; // Typically registered once when the queue worker boots (custom worker command, // AppServiceProvider, or a singleton bound in the container). $predictor = new WorkerHealthPredictor( memoryLimit: ini_get('memory_limit') ?: '128M', sampleWindowSize: 20, ); Event::listen(JobProcessed::class, function () use ($predictor): void { $predictor->recordSample(); $remaining = $predictor->predictRemainingJobs(); if ($remaining !== null && $remaining < 50) { // Ask the worker to exit after the current job so Supervisor/Octane restarts it. app('queue.worker')->shouldQuit = true; } });
For queue depth / TTO, feed Redis (or your driver) metrics on a schedule:
use Illuminate\Support\Facades\Redis; use PhpQueueProphet\QueueOverflowPredictor; $tracker = new QueueOverflowPredictor(maxCapacity: 50_000); $depth = (int) Redis::llen('queues:default'); // Derive rates from your own counters / Horizon metrics / time deltas. $tracker->record($depth, arrivalRate: 40.0, processingRate: 35.0); if ($tracker->getMetrics()->isAtRisk(120)) { logger()->warning('Queue nearing capacity', (array) $tracker->getMetrics()); }
Symfony Messenger
Use a middleware (or an event subscriber on WorkerRunningEvent / WorkerMessageHandledEvent):
use PhpQueueProphet\WorkerHealthPredictor; use Symfony\Component\Messenger\Envelope; use Symfony\Component\Messenger\Middleware\MiddlewareInterface; use Symfony\Component\Messenger\Middleware\StackInterface; use Symfony\Component\Messenger\Stamp\HandledStamp; final class ProphetMemoryMiddleware implements MiddlewareInterface { public function __construct( private readonly WorkerHealthPredictor $predictor, private readonly float $stopBelowJobs = 50.0, ) {} public function handle(Envelope $envelope, StackInterface $stack): Envelope { $envelope = $stack->next()->handle($envelope, $stack); if ($envelope->last(HandledStamp::class) === null) { return $envelope; } $this->predictor->recordSample(); $remaining = $this->predictor->predictRemainingJobs(); if ($remaining !== null && $remaining < $this->stopBelowJobs) { // Signal your worker loop / Messenger stop strategy to shut down gracefully. throw new \Symfony\Component\Messenger\Exception\StopWorkerException( 'Worker approaching OOM according to php-queue-prophet' ); } return $envelope; } }
Register the predictor as a shared service so the sliding window survives across messages in the same process:
# config/services.yaml PhpQueueProphet\WorkerHealthPredictor: arguments: $memoryLimit: '%env(default::memory_limit:MESSENGER_MEMORY_LIMIT)%' $sampleWindowSize: 20
How it works
Memory leak slope
Given (N) samples in the sliding window, where (x_i) is the sample index and (y_i) is memory in bytes:
[ m = \frac{N\sum(xy) - \sum x\sum y}{N\sum(x^2) - (\sum x)^2} ]
- (m \le 0) → stable or shrinking →
INF(no OOM risk from leakage) - (m > 0) → remaining jobs (\approx (\text{limit} - \text{current}) / m)
Queue TTO
[ \Delta = \text{arrival} - \text{processing},\quad \text{TTO} = \frac{\text{capacity} - \text{depth}}{\Delta} ]
- (\Delta \le 0) → draining/stable →
INF - Already at/over capacity →
0.0
API overview
| Class | Role |
|---|---|
WorkerHealthPredictor |
Sliding-window memory leak predictor |
QueueOverflowPredictor |
Queue capacity overflow estimator |
Support\LinearRegression |
OLS slope, intercept, R² |
DTO\HealthReport / DTO\QueueMetrics |
Immutable snapshots |
Contracts live under PhpQueueProphet\Contracts for easy mocking and DI.
Development
composer install composer test # PHPUnit composer test-coverage # PHPUnit + coverage (requires pcov or xdebug) composer phpstan # Level 8 composer cs-check # PER Coding Style 2.0 composer check # cs-check + phpstan + test
See CONTRIBUTING.md for PR guidelines and SECURITY.md for vulnerability reporting.
CI runs on PHP 8.1, 8.2, 8.3, and 8.4.
License
MIT © Fidel Alejandro Fernandez Arias