19694331639

Committed 26 Nov 2025 06:11AM UTC coverage: 87.651% (+5.0%) from 82.68%

Build # 19694331639

Build Type

push

github

Committed by

me-viper

Commit Message

test: Fix test concurrency

Run Details

4585 of 5231 relevant lines covered (87.65%)

0.88 hits per line

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

95.89

/src/Extensions.AspNetCore/src/PooledOpaPolicyService.cs

﻿using DotNext.Threading;

using Microsoft.Extensions.ObjectPool;
using Microsoft.Extensions.Options;
using Microsoft.Extensions.Primitives;

using OpaDotNet.Extensions.AspNetCore.Telemetry;
using OpaDotNet.Wasm;

namespace OpaDotNet.Extensions.AspNetCore;

internal class PooledOpaPolicyService : IOpaPolicyService, IDisposable
{
    private ObjectPool<IOpaEvaluator> _evaluatorPool;

    private readonly ILogger _logger;

    private readonly IDisposable _recompilationMonitor;

    private readonly OpaEvaluatorPoolProvider _poolProvider;

    private readonly IOpaPolicySource _factoryProvider;

    private readonly AsyncReaderWriterLock _syncLock = new();

    private readonly SemaphoreSlim? _concurrencyLock;

    public PooledOpaPolicyService(
        IOpaPolicySource factoryProvider,
        IOptions<OpaAuthorizationOptions> options,
        OpaEvaluatorPoolProvider poolProvider,
        ILogger<PooledOpaPolicyService> logger)
    {
        ArgumentNullException.ThrowIfNull(factoryProvider);
        ArgumentNullException.ThrowIfNull(options);
        ArgumentNullException.ThrowIfNull(poolProvider);
        ArgumentNullException.ThrowIfNull(logger);

        _factoryProvider = factoryProvider;
        _poolProvider = poolProvider;
        _logger = logger;

        _poolProvider.MaximumRetained = options.Value.MaximumEvaluatorsRetained;

        if (options.Value.MaximumEvaluators > 0)
            _concurrencyLock = new(options.Value.MaximumEvaluators, options.Value.MaximumEvaluators);

        _evaluatorPool = _poolProvider.Create(new OpaEvaluatorPoolPolicy(() => _factoryProvider.CreateEvaluator()));
        _recompilationMonitor = ChangeToken.OnChange(factoryProvider.OnPolicyUpdated, ResetPool);
    }

    private void ResetPool()
    {
        _logger.EvaluatorPoolResetting();

        var gotLock = _syncLock.TryEnterWriteLock(TimeSpan.FromSeconds(60));

        if (!gotLock)
            throw new TimeoutException("Failed to enter write lock within 60 seconds timeout");

        try
        {
            var oldPool = _evaluatorPool;
            _evaluatorPool = _poolProvider.Create(new OpaEvaluatorPoolPolicy(() => _factoryProvider.CreateEvaluator()));

            if (oldPool is not IDisposable pool)
            {
                _logger.EvaluatorPoolNotDisposable();
                return;
            }


            _logger.EvaluatorPoolDisposing();
            pool.Dispose();
        }
        finally
        {
            _syncLock.Release();
        }
    }

    private Task ConcurrencyLockWaitAsync(CancellationToken cancellationToken)
        => _concurrencyLock?.WaitAsync(cancellationToken) ?? Task.CompletedTask;

    public async ValueTask<bool> EvaluatePredicate<T>(T input, string entrypoint, CancellationToken cancellationToken)
    {
        ArgumentException.ThrowIfNullOrEmpty(entrypoint);

        await _syncLock.EnterReadLockAsync(cancellationToken).ConfigureAwait(false);
        await ConcurrencyLockWaitAsync(cancellationToken).ConfigureAwait(false);

        var pool = _evaluatorPool;
        var evaluator = pool.Get();

        try
        {
            var result = evaluator.EvaluatePredicate(input, entrypoint);
            return result.Result;
        }
        finally
        {
            pool.Return(evaluator);
            _concurrencyLock?.Release();
            _syncLock.Release();
        }
    }

    public async ValueTask<TOutput> Evaluate<TInput, TOutput>(TInput input, string entrypoint, CancellationToken cancellationToken)
        where TOutput : notnull
    {
        ArgumentException.ThrowIfNullOrEmpty(entrypoint);

        await _syncLock.EnterReadLockAsync(cancellationToken).ConfigureAwait(false);
        await ConcurrencyLockWaitAsync(cancellationToken).ConfigureAwait(false);

        var pool = _evaluatorPool;
        var evaluator = pool.Get();

        try
        {
            var result = evaluator.Evaluate<TInput, TOutput>(input, entrypoint);
            return result.Result;
        }
        finally
        {
            pool.Return(evaluator);
            _concurrencyLock?.Release();
            _syncLock.Release();
        }
    }

    public async ValueTask<string> EvaluateRaw(ReadOnlyMemory<char> inputJson, string entrypoint, CancellationToken cancellationToken)
    {
        ArgumentException.ThrowIfNullOrEmpty(entrypoint);

        await _syncLock.EnterReadLockAsync(cancellationToken).ConfigureAwait(false);
        await ConcurrencyLockWaitAsync(cancellationToken).ConfigureAwait(false);

        var pool = _evaluatorPool;
        var evaluator = pool.Get();

        try
        {
            return evaluator.EvaluateRaw(inputJson.Span, entrypoint);
        }
        finally
        {
            pool.Return(evaluator);
            _concurrencyLock?.Release();
            _syncLock.Release();
        }
    }

    public void Dispose()
    {
        _recompilationMonitor.Dispose();

        if (_evaluatorPool is IDisposable d)
            d.Dispose();

        _concurrencyLock?.Dispose();
        _syncLock.Dispose();
    }
}

1	using DotNext.Threading;
2
3	using Microsoft.Extensions.ObjectPool;
4	using Microsoft.Extensions.Options;
5	using Microsoft.Extensions.Primitives;
6
7	using OpaDotNet.Extensions.AspNetCore.Telemetry;
8	using OpaDotNet.Wasm;
9
10	namespace OpaDotNet.Extensions.AspNetCore;
11
12	internal class PooledOpaPolicyService : IOpaPolicyService, IDisposable
13	{
14	private ObjectPool<IOpaEvaluator> _evaluatorPool;
15
16	private readonly ILogger _logger;
17
18	private readonly IDisposable _recompilationMonitor;
19
20	private readonly OpaEvaluatorPoolProvider _poolProvider;
21
22	private readonly IOpaPolicySource _factoryProvider;
23
24	private readonly AsyncReaderWriterLock _syncLock = new();	1✔
25
26	private readonly SemaphoreSlim? _concurrencyLock;
27
28	public PooledOpaPolicyService(	1✔
29	IOpaPolicySource factoryProvider,	1✔
30	IOptions<OpaAuthorizationOptions> options,	1✔
31	OpaEvaluatorPoolProvider poolProvider,	1✔
32	ILogger<PooledOpaPolicyService> logger)	1✔
33	{
34	ArgumentNullException.ThrowIfNull(factoryProvider);	1✔
35	ArgumentNullException.ThrowIfNull(options);	1✔
36	ArgumentNullException.ThrowIfNull(poolProvider);	1✔
37	ArgumentNullException.ThrowIfNull(logger);	1✔
38
39	_factoryProvider = factoryProvider;	1✔
40	_poolProvider = poolProvider;	1✔
41	_logger = logger;	1✔
42
43	_poolProvider.MaximumRetained = options.Value.MaximumEvaluatorsRetained;	1✔
44
45	if (options.Value.MaximumEvaluators > 0)	1✔
46	_concurrencyLock = new(options.Value.MaximumEvaluators, options.Value.MaximumEvaluators);	1✔
47
48	_evaluatorPool = _poolProvider.Create(new OpaEvaluatorPoolPolicy(() => _factoryProvider.CreateEvaluator()));	1✔
49	_recompilationMonitor = ChangeToken.OnChange(factoryProvider.OnPolicyUpdated, ResetPool);	1✔
50	}	1✔
51
52	private void ResetPool()
53	{
54	_logger.EvaluatorPoolResetting();	1✔
55
56	var gotLock = _syncLock.TryEnterWriteLock(TimeSpan.FromSeconds(60));	1✔
57
58	if (!gotLock)	1✔
59	throw new TimeoutException("Failed to enter write lock within 60 seconds timeout");	×
60
61	try
62	{
63	var oldPool = _evaluatorPool;	1✔
64	_evaluatorPool = _poolProvider.Create(new OpaEvaluatorPoolPolicy(() => _factoryProvider.CreateEvaluator()));	1✔
65
66	if (oldPool is not IDisposable pool)	1✔
67	{
68	_logger.EvaluatorPoolNotDisposable();	×
69	return;	×
70	}
71
72
73	_logger.EvaluatorPoolDisposing();	1✔
74	pool.Dispose();	1✔
75	}	1✔
76	finally
77	{
78	_syncLock.Release();	1✔
79	}	1✔
80	}	1✔
81
82	private Task ConcurrencyLockWaitAsync(CancellationToken cancellationToken)
83	=> _concurrencyLock?.WaitAsync(cancellationToken) ?? Task.CompletedTask;	1✔
84
85	public async ValueTask<bool> EvaluatePredicate<T>(T input, string entrypoint, CancellationToken cancellationToken)
86	{
87	ArgumentException.ThrowIfNullOrEmpty(entrypoint);	1✔
88
89	await _syncLock.EnterReadLockAsync(cancellationToken).ConfigureAwait(false);	1✔
90	await ConcurrencyLockWaitAsync(cancellationToken).ConfigureAwait(false);	1✔
91
92	var pool = _evaluatorPool;	1✔
93	var evaluator = pool.Get();	1✔
94
95	try
96	{
97	var result = evaluator.EvaluatePredicate(input, entrypoint);	1✔
98	return result.Result;	1✔
99	}
100	finally
101	{
102	pool.Return(evaluator);	1✔
103	_concurrencyLock?.Release();	1✔
104	_syncLock.Release();	1✔
105	}
106	}	1✔
107
108	public async ValueTask<TOutput> Evaluate<TInput, TOutput>(TInput input, string entrypoint, CancellationToken cancellationToken)
109	where TOutput : notnull
110	{
111	ArgumentException.ThrowIfNullOrEmpty(entrypoint);	1✔
112
113	await _syncLock.EnterReadLockAsync(cancellationToken).ConfigureAwait(false);	1✔
114	await ConcurrencyLockWaitAsync(cancellationToken).ConfigureAwait(false);	1✔
115
116	var pool = _evaluatorPool;	1✔
117	var evaluator = pool.Get();	1✔
118
119	try
120	{
121	var result = evaluator.Evaluate<TInput, TOutput>(input, entrypoint);	1✔
122	return result.Result;	1✔
123	}
124	finally
125	{
126	pool.Return(evaluator);	1✔
127	_concurrencyLock?.Release();	1✔
128	_syncLock.Release();	1✔
129	}
130	}	1✔
131
132	public async ValueTask<string> EvaluateRaw(ReadOnlyMemory<char> inputJson, string entrypoint, CancellationToken cancellationToken)
133	{
134	ArgumentException.ThrowIfNullOrEmpty(entrypoint);	1✔
135
136	await _syncLock.EnterReadLockAsync(cancellationToken).ConfigureAwait(false);	1✔
137	await ConcurrencyLockWaitAsync(cancellationToken).ConfigureAwait(false);	1✔
138
139	var pool = _evaluatorPool;	1✔
140	var evaluator = pool.Get();	1✔
141
142	try
143	{
144	return evaluator.EvaluateRaw(inputJson.Span, entrypoint);	1✔
145	}
146	finally
147	{
148	pool.Return(evaluator);	1✔
149	_concurrencyLock?.Release();	1✔
150	_syncLock.Release();	1✔
151	}
152	}	1✔
153
154	public void Dispose()
155	{
156	_recompilationMonitor.Dispose();	1✔
157
158	if (_evaluatorPool is IDisposable d)	1✔
159	d.Dispose();	1✔
160
161	_concurrencyLock?.Dispose();	1✔
162	_syncLock.Dispose();	1✔
163	}	1✔
164	}

me-viper / OpaDotNet / 19694331639

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