24850441053

Committed 23 Apr 2026 05:54PM UTC coverage: 52.404% (-1.5%) from 53.938%

Build # 24850441053

Build Type

push

github

Committed by

daycry

Commit Message

Fixes

Coverage Stats

104 of 219 new or added lines in 42 files covered. (47.49%)

14 existing lines in 9 files now uncovered.

1210 of 2309 relevant lines covered (52.4%)

4.37 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/Libraries/RateLimiter.php

<?php

declare(strict_types=1);

/**
 * This file is part of Daycry Queues.
 *
 * (c) Daycry <daycry9@proton.me>
 *
 * For the full copyright and license information, please view
 * the LICENSE file that was distributed with this source code.
 */

namespace Daycry\Jobs\Libraries;

use Daycry\Jobs\Exceptions\JobException;

/**
 * Rate limiter for queue processing using cache-based token bucket algorithm.
 * Prevents queue saturation by limiting jobs processed per minute.
 */
class RateLimiter
{
    /**
     * Check if a queue can process another job (rate limit check).
     *
     * @param string $queue        Queue name
     * @param int    $maxPerMinute Maximum jobs allowed per minute (0 = unlimited)
     *
     * @return bool True if allowed, false if throttled
     */
    public function allow(string $queue, int $maxPerMinute): bool
    {
        if ($maxPerMinute <= 0) {
            return true; // Unlimited
        }

        $key   = "queue_rate_{$queue}";
        $cache = service('cache');

        // Attempt atomic increment if the cache driver supports it (e.g. Redis, Memcached).
        // This prevents race conditions between get() and save() calls.
        if (method_exists($cache, 'increment')) {
            $current = $cache->get($key);

            if ($current === null) {
                // First request in this window: initialize counter
                $cache->save($key, 1, 60);

                return true;
            }

            if ((int) $current >= $maxPerMinute) {
                return false; // Throttled
            }

            $cache->increment($key, 1);

            return true;
        }

        // Fallback for cache drivers without atomic increment
        $count = (int) ($cache->get($key) ?? 0);

        if ($count >= $maxPerMinute) {
            return false; // Throttled
        }

        // Increment counter (expires after 60 seconds)
        $cache->save($key, $count + 1, 60);

        return true;
    }

    /**
     * Throw exception if rate limit exceeded.
     *
     * @throws JobException
     */
    public function throttle(string $queue, int $maxPerMinute): void
    {
        if (! $this->allow($queue, $maxPerMinute)) {
            throw JobException::forRateLimitExceeded($queue, $maxPerMinute);
        }
    }

    /**
     * Get current rate usage for a queue.
     *
     * @return int Number of jobs processed in current minute
     */
    public function getUsage(string $queue): int
    {
        $key   = "queue_rate_{$queue}";
        $cache = service('cache');

        return (int) ($cache->get($key) ?? 0);
    }

    /**
     * Reset rate limit counter for a queue.
     */
    public function reset(string $queue): void
    {
        $key   = "queue_rate_{$queue}";
        $cache = service('cache');
        $cache->delete($key);
    }
}

1	<?php
2
3	declare(strict_types=1);
4
5	/**
6	* This file is part of Daycry Queues.
7	*
8	* (c) Daycry <daycry9@proton.me>
9	*
10	* For the full copyright and license information, please view
11	* the LICENSE file that was distributed with this source code.
12	*/
13
14	namespace Daycry\Jobs\Libraries;
15
16	use Daycry\Jobs\Exceptions\JobException;
17
18	/**
19	* Rate limiter for queue processing using cache-based token bucket algorithm.
20	* Prevents queue saturation by limiting jobs processed per minute.
21	*/
22	class RateLimiter
23	{
24	/**
25	* Check if a queue can process another job (rate limit check).
26	*
27	* @param string $queue Queue name
28	* @param int $maxPerMinute Maximum jobs allowed per minute (0 = unlimited)
29	*
30	* @return bool True if allowed, false if throttled
31	*/
32	public function allow(string $queue, int $maxPerMinute): bool
33	{
34	if ($maxPerMinute <= 0) {	×
35	return true; // Unlimited	×
36	}
37
38	$key = "queue_rate_{$queue}";	×
39	$cache = service('cache');	×
40
41	// Attempt atomic increment if the cache driver supports it (e.g. Redis, Memcached).
42	// This prevents race conditions between get() and save() calls.
NEW 43	if (method_exists($cache, 'increment')) {	×
NEW 44	$current = $cache->get($key);	×
45
NEW 46	if ($current === null) {	×
47	// First request in this window: initialize counter
NEW 48	$cache->save($key, 1, 60);	×
49
NEW 50	return true;	×
51	}
52
NEW 53	if ((int) $current >= $maxPerMinute) {	×
NEW 54	return false; // Throttled	×
55	}
56
NEW 57	$cache->increment($key, 1);	×
58
NEW 59	return true;	×
60	}
61
62	// Fallback for cache drivers without atomic increment
UNCOV 63	$count = (int) ($cache->get($key) ?? 0);	×
64
65	if ($count >= $maxPerMinute) {	×
66	return false; // Throttled	×
67	}
68
69	// Increment counter (expires after 60 seconds)
70	$cache->save($key, $count + 1, 60);	×
71
72	return true;	×
73	}
74
75	/**
76	* Throw exception if rate limit exceeded.
77	*
78	* @throws JobException
79	*/
80	public function throttle(string $queue, int $maxPerMinute): void
81	{
82	if (! $this->allow($queue, $maxPerMinute)) {	×
83	throw JobException::forRateLimitExceeded($queue, $maxPerMinute);	×
84	}
85	}
86
87	/**
88	* Get current rate usage for a queue.
89	*
90	* @return int Number of jobs processed in current minute
91	*/
92	public function getUsage(string $queue): int
93	{
94	$key = "queue_rate_{$queue}";	×
95	$cache = service('cache');	×
96
97	return (int) ($cache->get($key) ?? 0);	×
98	}
99
100	/**
101	* Reset rate limit counter for a queue.
102	*/
103	public function reset(string $queue): void
104	{
105	$key = "queue_rate_{$queue}";	×
106	$cache = service('cache');	×
107	$cache->delete($key);	×
108	}
109	}

daycry / jobs / 24850441053

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous