[12.x] Add Cache::funnel() for concurrency limiting with any cache driver

[mathiasgrimm]

• Enables concurrency limiting with any cache driver, not just Redis — helpful for testing (where array is commonly used) and for environments using database or other drivers.
• Allows easier transition between cache drivers without code changes.

Usage

// With Cache (any driver)
Cache::funnel($key)
    ->limit(5)
    ->releaseAfter(60)
    ->then(function () {
        // do something
    }, function () {
        // could not get the lock
    });

// Redis::funnel() continues to work as before
Redis::funnel($key)
    ->limit(5)
    ->releaseAfter(60)
    ->then(function () {
        // do something
    }, function () {
        // could not get the lock
    });

Changes

Extracts concurrency limiter functionality from Redis-specific implementation to a cache-driver-agnostic base in Illuminate\Cache.

• Add funnel() method to Cache Repository
• Create base ConcurrencyLimiter and ConcurrencyLimiterBuilder classes in Illuminate\Cache\Limiters
• Create LimiterTimeoutException in Illuminate\Cache\Limiters
• Refactor Redis limiters to extend new base classes
• Redis ConcurrencyLimiter reuses parent's block() via newTimeoutException() factory method
• Redis ConcurrencyLimiterBuilder reuses parent's then() since exception hierarchy allows it

Design Decisions

Redis ConcurrencyLimiter doesn't call parent::__construct()

The Redis subclass has a fundamentally different dependency (Connection vs LockProvider). This is consistent with how cache stores like RedisStore, MemcachedStore, and ArrayStore extend TaggableStore without calling parent constructor.

block() is shared via newTimeoutException() factory method

The base and Redis block() methods were identical except for which exception they throw. A newTimeoutException() factory method (overridden by Redis) eliminates the duplication. This follows the framework's established factory method pattern (newHasOne(), newMorphTo(), newPendingRequest(), etc.).

No LockProvider guard on funnel()

Consistent with the existing withoutOverlapping() method which also calls $store->lock() without checking. All built-in cache stores implement LockProvider.

DurationLimiter not extracted

Kept out of scope — Cache::throttle() can be a follow-up PR.

Manual Testing

This is covered by the Integration tests, but could be useful for some extra manual testing.

Route::get('/funnel-tests', function () {
    $results = [];
    $pass = function ($name) use (&$results) { $results[] = ['test' => $name, 'status' => 'PASS']; };
    $fail = function ($name, $reason = '') use (&$results) { $results[] = ['test' => $name, 'status' => 'FAIL', 'reason' => $reason]; };

    // ---------------------------------------------------------------
    // Drivers to test: array, file, database (mysql), redis
    // ---------------------------------------------------------------
    $drivers = ['array', 'file', 'database', 'redis'];

    foreach ($drivers as $driver) {
        $store = Cache::store($driver);

        // flush() clears cache data but NOT locks (separate table/connection),
        // so we also force-release every lock key the tests will use.
        $store->flush();
        $lockProvider = $store->getStore();
        $testKeys = ['basic', 'seq', 'overflow', 'exc', 'timeout', 'indep-a', 'indep-b'];
        foreach ($testKeys as $name) {
            for ($slot = 1; $slot <= 2; $slot++) {
                $lockProvider->lock("{$driver}-{$name}{$slot}")->forceRelease();
            }
        }

        $label = "[{$driver}]";
        $key = fn ($name) => "{$driver}-{$name}";

        // 1. Basic happy path
        try {
            $val = $store->funnel($key('basic'))
                ->limit(2)->releaseAfter(10)->block(3)
                ->then(fn () => 'hello', fn () => null);
            $val === 'hello'
                ? $pass("{$label} basic happy path")
                : $fail("{$label} basic happy path", "expected 'hello', got '{$val}'");
        } catch (Throwable $e) {
            $fail("{$label} basic happy path", $e->getMessage());
        }

        // 2. Sequential runs – lock released after each callback
        try {
            $ok = true;
            foreach (range(1, 5) as $i) {
                $r = $store->funnel($key('seq'))->limit(1)->releaseAfter(10)->block(2)
                    ->then(fn () => 'ok', fn () => 'blocked');
                if ($r !== 'ok') { $ok = false; break; }
            }
            $ok
                ? $pass("{$label} sequential (5 runs, limit=1)")
                : $fail("{$label} sequential", "run #{$i} was blocked");
        } catch (Throwable $e) {
            $fail("{$label} sequential", $e->getMessage());
        }

        // 3. Overflow – 3rd acquire fails when limit=2 and 2 slots held
        try {
            $lp = $store->getStore();
            $held = new class($lp, $key('overflow'), 2, 60) extends ConcurrencyLimiter {
                protected function release($lock, $id) {}
            };
            $held->block(1, fn () => null);
            $held->block(1, fn () => null);

            $r = $store->funnel($key('overflow'))->limit(2)->releaseAfter(60)->block(0)
                ->then(fn () => 'acquired', fn () => 'blocked');
            $r === 'blocked'
                ? $pass("{$label} overflow (limit=2, 3rd blocked)")
                : $fail("{$label} overflow", 'should have been blocked');
        } catch (Throwable $e) {
            $fail("{$label} overflow", $e->getMessage());
        }

        // 4. Lock released on exception
        try {
            try {
                $store->funnel($key('exc'))->limit(1)->releaseAfter(60)->block(2)
                    ->then(fn () => throw new RuntimeException('Boom'));
            } catch (RuntimeException) {}

            $r = $store->funnel($key('exc'))->limit(1)->releaseAfter(60)->block(0)
                ->then(fn () => 'acquired', fn () => 'still-locked');
            $r === 'acquired'
                ? $pass("{$label} exception releases lock")
                : $fail("{$label} exception releases lock", 'lock was not released');
        } catch (Throwable $e) {
            $fail("{$label} exception releases lock", $e->getMessage());
        }

        // 5. LimiterTimeoutException thrown without failure callback
        try {
            $lp = $store->getStore();
            $held = new class($lp, $key('timeout'), 1, 60) extends ConcurrencyLimiter {
                protected function release($lock, $id) {}
            };
            $held->block(1, fn () => null);

            try {
                $store->funnel($key('timeout'))->limit(1)->releaseAfter(60)->block(0)
                    ->then(fn () => 'nope');
                $fail("{$label} timeout exception", 'no exception thrown');
            } catch (LimiterTimeoutException) {
                $pass("{$label} timeout exception");
            }
        } catch (Throwable $e) {
            $fail("{$label} timeout exception", $e->getMessage());
        }

        // 6. Independent keys
        try {
            $lp = $store->getStore();
            $held = new class($lp, $key('indep-a'), 1, 60) extends ConcurrencyLimiter {
                protected function release($lock, $id) {}
            };
            $held->block(1, fn () => null);

            $r = $store->funnel($key('indep-b'))->limit(1)->releaseAfter(60)->block(0)
                ->then(fn () => 'acquired', fn () => 'blocked');
            $r === 'acquired'
                ? $pass("{$label} independent keys")
                : $fail("{$label} independent keys", 'key-b blocked by key-a');
        } catch (Throwable $e) {
            $fail("{$label} independent keys", $e->getMessage());
        }
    }

    // Render results
    $total = count($results);
    $passed = count(array_filter($results, fn ($r) => $r['status'] === 'PASS'));
    $lines = ["Cache::funnel() test results ({$passed}/{$total} passed)", str_repeat('=', 60)];

    $currentDriver = '';
    foreach ($results as $r) {
        // Extract driver name from "[driver] test name"
        if (preg_match('/^\[(\w+)\]/', $r['test'], $m) && $m[1] !== $currentDriver) {
            $currentDriver = $m[1];
            $lines[] = '';
            $lines[] = "--- {$currentDriver} " . str_repeat('-', 55 - strlen($currentDriver));
        }
        $icon = $r['status'] === 'PASS' ? '✅' : '❌';
        $line = "  {$icon} [{$r['status']}] {$r['test']}";
        if (!empty($r['reason'])) $line .= " — {$r['reason']}";
        $lines[] = $line;
    }
    $lines[] = '';
    $lines[] = str_repeat('=', 60);
    $lines[] = $passed === $total ? '✅ ALL TESTS PASSED' : '❌ SOME TESTS FAILED';

    return response(implode("\n", $lines), 200, ['Content-Type' => 'text/plain']);
});

[taylorotwell]

I don't think this should extend something from the cache component. I would leave it as a separate exception.

[taylorotwell]

This should probably make sure the cache actually supports locks.

[taylorotwell]

I don't think this should extend the cache component stuff.

[taylorotwell]

I think this is a cool feature but I would just make it separate from the Redis stuff even though there will be duplication. 👍