natsdotnet/tests/NATS.Server.JetStream.Tests/JetStream/Api/ApiRateLimiterTests.cs

// Go reference: jetstream_api.go — rate limiting via maxConcurrentRequests semaphore and
// request deduplication via the dedup cache keyed by Nats-Msg-Id header.
// The Go server uses a configurable semaphore (default 256) to throttle concurrent API
// requests, and caches responses for duplicate request IDs within a TTL window.

namespace NATS.Server.JetStream.Tests.JetStream.Api;

using NATS.Server.JetStream.Api;

public class ApiRateLimiterTests : IDisposable
{
    private readonly ApiRateLimiter _limiter = new(maxConcurrent: 4);

    public void Dispose() => _limiter.Dispose();

    // Go reference: jetstream_api.go — semaphore.TryAcquire(0) used for non-blocking attempt.
    [Fact]
    public async Task TryAcquire_succeeds_when_slots_available()
    {
        var acquired = await _limiter.TryAcquireAsync();
        acquired.ShouldBeTrue();
    }

    // Go reference: jetstream_api.go — when all slots are taken, new requests are rejected.
    [Fact]
    public async Task TryAcquire_fails_when_all_slots_taken()
    {
        // Fill all 4 slots.
        for (var i = 0; i < 4; i++)
            (await _limiter.TryAcquireAsync()).ShouldBeTrue();

        // 5th attempt should fail.
        var rejected = await _limiter.TryAcquireAsync();
        rejected.ShouldBeFalse();
    }

    // Go reference: jetstream_api.go — releasing a slot allows a subsequent request to proceed.
    [Fact]
    public async Task Release_frees_slot_for_next_request()
    {
        // Fill all slots.
        for (var i = 0; i < 4; i++)
            (await _limiter.TryAcquireAsync()).ShouldBeTrue();

        // Currently full.
        (await _limiter.TryAcquireAsync()).ShouldBeFalse();

        // Release one slot.
        _limiter.Release();

        // Now one slot is free.
        (await _limiter.TryAcquireAsync()).ShouldBeTrue();
    }

    // Go reference: jetstream_api.go — active count reflects in-flight requests.
    [Fact]
    public async Task ActiveCount_tracks_concurrent_requests()
    {
        _limiter.ActiveCount.ShouldBe(0);

        await _limiter.TryAcquireAsync();
        await _limiter.TryAcquireAsync();
        await _limiter.TryAcquireAsync();

        _limiter.ActiveCount.ShouldBe(3);
    }

    // Go reference: jetstream_api.go — unknown request ID returns null (cache miss).
    [Fact]
    public void GetCachedResponse_returns_null_for_unknown_id()
    {
        var result = _limiter.GetCachedResponse("nonexistent-id");
        result.ShouldBeNull();
    }

    // Go reference: jetstream_api.go — dedup cache stores response keyed by Nats-Msg-Id.
    [Fact]
    public void CacheResponse_and_get_returns_cached()
    {
        var response = JetStreamApiResponse.SuccessResponse();
        _limiter.CacheResponse("req-001", response);

        var cached = _limiter.GetCachedResponse("req-001");
        cached.ShouldNotBeNull();
        cached!.Success.ShouldBeTrue();
    }

    // Go reference: jetstream_api.go — dedup window expires after TTL (dedupWindow config).
    [SlopwatchSuppress("SW004", "TTL expiry test requires real wall-clock time to elapse; no synchronisation primitive can replace observing a time-based cache eviction")]
    [Fact]
    public async Task GetCachedResponse_returns_null_after_ttl_expiry()
    {
        using var shortLimiter = new ApiRateLimiter(maxConcurrent: 4, dedupTtl: TimeSpan.FromMilliseconds(50));
        var response = JetStreamApiResponse.SuccessResponse();
        shortLimiter.CacheResponse("req-ttl", response);

        // Verify it's cached before expiry.
        shortLimiter.GetCachedResponse("req-ttl").ShouldNotBeNull();

        // Wait for TTL to expire.
        await Task.Delay(120);

        // Should be null after expiry.
        shortLimiter.GetCachedResponse("req-ttl").ShouldBeNull();
    }

    // Go reference: jetstream_api.go — null/empty Nats-Msg-Id is ignored for dedup.
    [Fact]
    public void CacheResponse_ignores_null_request_id()
    {
        var response = JetStreamApiResponse.SuccessResponse();

        // These should not throw and should not increment the cache count.
        _limiter.CacheResponse(null, response);
        _limiter.CacheResponse("", response);
        _limiter.CacheResponse(string.Empty, response);

        _limiter.DedupCacheCount.ShouldBe(0);
        _limiter.GetCachedResponse(null).ShouldBeNull();
        _limiter.GetCachedResponse("").ShouldBeNull();
    }

    // Go reference: jetstream_api.go — periodic sweep removes expired dedup entries.
    [SlopwatchSuppress("SW004", "TTL expiry test requires real wall-clock time to elapse; no synchronisation primitive can replace observing a time-based cache eviction")]
    [Fact]
    public async Task PurgeExpired_removes_old_entries()
    {
        using var shortLimiter = new ApiRateLimiter(maxConcurrent: 4, dedupTtl: TimeSpan.FromMilliseconds(50));

        shortLimiter.CacheResponse("req-a", JetStreamApiResponse.SuccessResponse());
        shortLimiter.CacheResponse("req-b", JetStreamApiResponse.SuccessResponse());
        shortLimiter.CacheResponse("req-c", JetStreamApiResponse.SuccessResponse());

        shortLimiter.DedupCacheCount.ShouldBe(3);

        // Wait for all entries to expire.
        await Task.Delay(120);

        var removed = shortLimiter.PurgeExpired();
        removed.ShouldBe(3);
        shortLimiter.DedupCacheCount.ShouldBe(0);
    }

    // Go reference: jetstream_api.go — dedup cache count is observable.
    [Fact]
    public void DedupCacheCount_tracks_cached_entries()
    {
        _limiter.DedupCacheCount.ShouldBe(0);

        _limiter.CacheResponse("req-1", JetStreamApiResponse.Ok());
        _limiter.CacheResponse("req-2", JetStreamApiResponse.Ok());
        _limiter.CacheResponse("req-3", JetStreamApiResponse.Ok());

        _limiter.DedupCacheCount.ShouldBe(3);
    }

    // Go reference: jetstream_api.go — semaphore enforces max-concurrent across goroutines.
    [Fact]
    public async Task Concurrent_acquire_respects_max()
    {
        using var limiter = new ApiRateLimiter(maxConcurrent: 5);

        // Spin up 10 tasks, only 5 should succeed.
        var results = await Task.WhenAll(
            Enumerable.Range(0, 10).Select(_ => limiter.TryAcquireAsync()));

        var acquired = results.Count(r => r);
        acquired.ShouldBe(5);
    }

    // Go reference: jetstream_api.go — default maxConcurrentRequests = 256.
    [Fact]
    public async Task Default_max_concurrent_is_256()
    {
        using var defaultLimiter = new ApiRateLimiter();

        // Acquire 256 slots — all should succeed.
        var tasks = Enumerable.Range(0, 256).Select(_ => defaultLimiter.TryAcquireAsync());
        var results = await Task.WhenAll(tasks);
        results.ShouldAllBe(r => r);

        // 257th should fail.
        var rejected = await defaultLimiter.TryAcquireAsync();
        rejected.ShouldBeFalse();
    }
}