using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Jellyfin.Database.Implementations;
using Jellyfin.Database.Implementations.DbConfiguration;
using Jellyfin.Database.Implementations.Entities;
using Jellyfin.Database.Implementations.Locking;
using Jellyfin.Database.Providers.PostgreSQL;
using Jellyfin.Server.Tests.Migrations;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging.Abstractions;
using Npgsql;
using Xunit;
namespace Jellyfin.Database.Tests.PostgreSQL;
///
/// Integration tests that verify concurrent access patterns against a real PostgreSQL server.
///
[Xunit.Trait("Category", "RequiresDocker")]
public sealed class PostgreSqlConcurrencyTests : IAsyncLifetime
{
private static int _databaseSequence;
private PostgreSqlTestServer? _server;
private NpgsqlDataSource? _dataSource;
private PostgreSqlDatabaseProvider? _provider;
///
/// Attaches to the test server, hands this test a database of its own and applies migrations to it.
///
/// A representing the asynchronous operation.
public async ValueTask InitializeAsync()
{
_server = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false);
var databaseName = FormattableString.Invariant($"pg_concurrency_{Interlocked.Increment(ref _databaseSequence)}");
var connectionString = await _server.CreateDatabaseAsync(databaseName, TestContext.Current.CancellationToken).ConfigureAwait(false);
_dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
_provider = new PostgreSqlDatabaseProvider(_dataSource);
var context = CreateContext();
await using (context.ConfigureAwait(false))
{
await context.Database.MigrateAsync(TestContext.Current.CancellationToken).ConfigureAwait(false);
}
}
///
/// Releases the data source and the test server.
///
/// A representing the asynchronous operation.
public async ValueTask DisposeAsync()
{
// InitializeAsync can fail before the data source exists; its error must not be masked by an NRE here.
if (_dataSource is not null)
{
await _dataSource.DisposeAsync().ConfigureAwait(false);
}
if (_server is not null)
{
await _server.DisposeAsync().ConfigureAwait(false);
}
}
///
/// Verifies that concurrent inserts on from four parallel tasks succeed without deadlock.
///
/// A representing the asynchronous operation.
[Fact]
public async Task ConcurrentInserts_ActivityLogs_SucceedWithoutDeadlock()
{
const int parallelTasks = 4;
const int insertsPerTask = 10;
var tasks = new List(parallelTasks);
for (var i = 0; i < parallelTasks; i++)
{
var taskIndex = i;
tasks.Add(Task.Run(
() => InsertBatchAsync(taskIndex, insertsPerTask),
TestContext.Current.CancellationToken));
}
await Task.WhenAll(tasks);
// Verify all rows were inserted
var verifyCtx = CreateContext();
await using (verifyCtx)
{
var count = await verifyCtx.ActivityLogs
.CountAsync(l => l.Type == "ConcurrencyTest", TestContext.Current.CancellationToken);
Assert.Equal(parallelTasks * insertsPerTask, count);
}
}
private async Task InsertBatchAsync(int taskIndex, int insertsPerTask)
{
var ctx = CreateContext();
await using (ctx.ConfigureAwait(false))
{
for (var j = 0; j < insertsPerTask; j++)
{
ctx.ActivityLogs.Add(new ActivityLog(
$"Task {taskIndex} Insert {j}",
"ConcurrencyTest",
Guid.Empty));
}
await ctx.SaveChangesAsync().ConfigureAwait(false);
}
}
private JellyfinDbContext CreateContext()
{
var optionsBuilder = new DbContextOptionsBuilder();
_provider!.Initialise(optionsBuilder, new DatabaseConfigurationOptions { DatabaseType = "PostgreSQL" });
return new JellyfinDbContext(
optionsBuilder.Options,
NullLogger.Instance,
_provider,
new NoLockBehavior(NullLogger.Instance));
}
}