test: create test for concurrency
This commit is contained in:
voltsrage
@@ -0,0 +1,534 @@
|
||||
using System.Text.Json;
|
||||
using Confluent.Kafka;
|
||||
using FluentAssertions;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using StackExchange.Redis;
|
||||
|
||||
[Collection("Integration")]
|
||||
public class BackgroundServiceTests : IAsyncLifetime
|
||||
{
|
||||
private readonly ApiFixture _fixture;
|
||||
|
||||
public BackgroundServiceTests(ApiFixture fixture) => _fixture = fixture;
|
||||
|
||||
public async Task InitializeAsync()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
await DbResetHelper.ResetAsync(db);
|
||||
|
||||
var redis = scope.ServiceProvider.GetRequiredService<IConnectionMultiplexer>();
|
||||
var server = redis.GetServer(redis.GetEndPoints().First());
|
||||
await server.FlushDatabaseAsync(1);
|
||||
}
|
||||
|
||||
public Task DisposeAsync() => Task.CompletedTask;
|
||||
|
||||
// =======================================================================
|
||||
// PoisonPillGuard — unit tests (no infrastructure needed)
|
||||
// =======================================================================
|
||||
|
||||
[Fact]
|
||||
public void PoisonPillGuard_PermanentError_SkipsImmediately()
|
||||
{
|
||||
var guard = new PoisonPillGuard("test-group", maxRetries: 5, NullLogger.Instance);
|
||||
var result = BuildConsumeResult("test-topic", 0, 0, "bad payload");
|
||||
|
||||
var shouldSkip = guard.ShouldSkip(result, new JsonException("Invalid JSON"));
|
||||
|
||||
shouldSkip.Should().BeTrue("permanent errors (JsonException) must be skipped immediately");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void PoisonPillGuard_FormatException_SkipsImmediately()
|
||||
{
|
||||
var guard = new PoisonPillGuard("test-group", maxRetries: 5, NullLogger.Instance);
|
||||
var result = BuildConsumeResult("test-topic", 0, 0, "not-a-guid");
|
||||
|
||||
var shouldSkip = guard.ShouldSkip(result, new FormatException("Input string was not in a correct format"));
|
||||
|
||||
shouldSkip.Should().BeTrue("FormatException is a permanent error and must be skipped immediately");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void PoisonPillGuard_WrappedPermanentError_SkipsImmediately()
|
||||
{
|
||||
var guard = new PoisonPillGuard("test-group", maxRetries: 5, NullLogger.Instance);
|
||||
var result = BuildConsumeResult("test-topic", 0, 0, "bad");
|
||||
|
||||
var wrappedException = new InvalidOperationException(
|
||||
"Processing failed", new JsonException("Invalid JSON"));
|
||||
|
||||
var shouldSkip = guard.ShouldSkip(result, wrappedException);
|
||||
|
||||
shouldSkip.Should().BeTrue(
|
||||
"permanent errors wrapped in other exceptions must still be detected via root cause");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void PoisonPillGuard_TransientError_RetriesUpToMax()
|
||||
{
|
||||
const int maxRetries = 3;
|
||||
var guard = new PoisonPillGuard("test-group", maxRetries, NullLogger.Instance);
|
||||
var result = BuildConsumeResult("test-topic", 0, 42, "payload");
|
||||
var transientError = new TimeoutException("Connection timed out");
|
||||
|
||||
for (var i = 1; i < maxRetries; i++)
|
||||
{
|
||||
var shouldSkip = guard.ShouldSkip(result, transientError);
|
||||
shouldSkip.Should().BeFalse($"retry {i}/{maxRetries} should not skip");
|
||||
}
|
||||
|
||||
var finalSkip = guard.ShouldSkip(result, transientError);
|
||||
finalSkip.Should().BeTrue(
|
||||
$"after {maxRetries} retries, the message must be skipped");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void PoisonPillGuard_SuccessResetsRetryCount()
|
||||
{
|
||||
const int maxRetries = 3;
|
||||
var guard = new PoisonPillGuard("test-group", maxRetries, NullLogger.Instance);
|
||||
var result = BuildConsumeResult("test-topic", 0, 42, "payload");
|
||||
var transientError = new TimeoutException("Connection timed out");
|
||||
|
||||
guard.ShouldSkip(result, transientError);
|
||||
guard.ShouldSkip(result, transientError);
|
||||
|
||||
guard.OnSuccess();
|
||||
|
||||
guard.ShouldSkip(result, transientError).Should().BeFalse(
|
||||
"after OnSuccess, retry count resets and first failure should not skip");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void PoisonPillGuard_DifferentOffset_ResetsRetryCount()
|
||||
{
|
||||
const int maxRetries = 3;
|
||||
var guard = new PoisonPillGuard("test-group", maxRetries, NullLogger.Instance);
|
||||
var transientError = new TimeoutException("Connection timed out");
|
||||
|
||||
var result1 = BuildConsumeResult("test-topic", 0, 42, "payload-a");
|
||||
guard.ShouldSkip(result1, transientError);
|
||||
guard.ShouldSkip(result1, transientError);
|
||||
|
||||
var result2 = BuildConsumeResult("test-topic", 0, 43, "payload-b");
|
||||
guard.ShouldSkip(result2, transientError).Should().BeFalse(
|
||||
"a new offset resets the retry counter");
|
||||
}
|
||||
|
||||
// =======================================================================
|
||||
// Outbox relay — retry count and dead-letter behavior
|
||||
// =======================================================================
|
||||
|
||||
[Fact]
|
||||
public async Task OutboxEvent_UnprocessedEvents_ArePickedUpByRelay()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var encounterId = Guid.NewGuid();
|
||||
db.OutboxEvents.Add(new OutboxEvent
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
Topic = "observation.recorded",
|
||||
Payload = JsonSerializer.Serialize(new { encounterId, code = "HEART_RATE", value = 80 }),
|
||||
PartitionKey = encounterId.ToString(),
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
});
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
var pendingBefore = await db.OutboxEvents
|
||||
.CountAsync(e => e.ProcessedAt == null && e.FailedAt == null);
|
||||
|
||||
pendingBefore.Should().BeGreaterThanOrEqualTo(1);
|
||||
|
||||
var deadline = DateTimeOffset.UtcNow.AddSeconds(15);
|
||||
while (DateTimeOffset.UtcNow < deadline)
|
||||
{
|
||||
await Task.Delay(500);
|
||||
var remaining = await db.OutboxEvents
|
||||
.AsNoTracking()
|
||||
.CountAsync(e => e.ProcessedAt == null && e.FailedAt == null);
|
||||
if (remaining == 0) break;
|
||||
}
|
||||
|
||||
var unprocessed = await db.OutboxEvents
|
||||
.AsNoTracking()
|
||||
.CountAsync(e => e.ProcessedAt == null && e.FailedAt == null);
|
||||
|
||||
unprocessed.Should().Be(0, "the outbox relay should process all pending events");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task OutboxEvent_FailedEvents_HaveRetryCountAndError()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var failedEvent = new OutboxEvent
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
Topic = "nonexistent.topic.that.will.fail",
|
||||
Payload = "{}",
|
||||
PartitionKey = "test",
|
||||
CreatedAt = DateTimeOffset.UtcNow,
|
||||
RetryCount = 9,
|
||||
LastError = "Simulated prior failure"
|
||||
};
|
||||
db.OutboxEvents.Add(failedEvent);
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
failedEvent.RetryCount.Should().Be(9);
|
||||
failedEvent.LastError.Should().NotBeNullOrEmpty();
|
||||
failedEvent.ProcessedAt.Should().BeNull();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task OutboxEvent_FailedAtSet_EventIsNotReprocessed()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var deadLetteredEvent = new OutboxEvent
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
Topic = "test.topic",
|
||||
Payload = "{}",
|
||||
PartitionKey = "test",
|
||||
CreatedAt = DateTimeOffset.UtcNow,
|
||||
RetryCount = 10,
|
||||
LastError = "Exceeded max retries",
|
||||
FailedAt = DateTimeOffset.UtcNow.AddMinutes(-1)
|
||||
};
|
||||
db.OutboxEvents.Add(deadLetteredEvent);
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
await Task.Delay(3_000);
|
||||
|
||||
var evt = await db.OutboxEvents
|
||||
.AsNoTracking()
|
||||
.SingleAsync(e => e.Id == deadLetteredEvent.Id);
|
||||
|
||||
evt.ProcessedAt.Should().BeNull("dead-lettered events must not be reprocessed");
|
||||
evt.FailedAt.Should().NotBeNull();
|
||||
}
|
||||
|
||||
// =======================================================================
|
||||
// SepsisBundleMonitor — concurrent scan safety
|
||||
// =======================================================================
|
||||
|
||||
[Fact]
|
||||
public async Task SepsisBundleMonitor_ConcurrentScans_NoDuplicateUpdates()
|
||||
{
|
||||
var (_, encounterId) = await SeedPatientEncounterAndBundleAsync();
|
||||
|
||||
using (var scope = _fixture.Services.CreateScope())
|
||||
{
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
var bundle = await db.SepsisBundles.SingleAsync(b => b.EncounterId == encounterId);
|
||||
bundle.DeadlineAt = DateTimeOffset.UtcNow.AddHours(-1);
|
||||
await db.SaveChangesAsync();
|
||||
}
|
||||
|
||||
var tasks = Enumerable.Range(0, 5).Select(_ =>
|
||||
{
|
||||
return Task.Run(async () =>
|
||||
{
|
||||
var monitor = new SepsisBundleMonitorService(
|
||||
_fixture.Services,
|
||||
_fixture.Services.GetRequiredService<ClinicalMetrics>(),
|
||||
_fixture.Services.GetRequiredService<ILogger<SepsisBundleMonitorService>>());
|
||||
await monitor.ScanOverdueBundlesAsync(CancellationToken.None);
|
||||
});
|
||||
});
|
||||
|
||||
await Task.WhenAll(tasks);
|
||||
|
||||
using var verifyScope = _fixture.Services.CreateScope();
|
||||
var verifyDb = verifyScope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var bundles = await verifyDb.SepsisBundles
|
||||
.AsNoTracking()
|
||||
.Where(b => b.EncounterId == encounterId)
|
||||
.ToListAsync();
|
||||
|
||||
bundles.Should().HaveCount(1);
|
||||
bundles[0].ComplianceStatus.Should().Be(SepsisBundleComplianceStatus.NonCompliant);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SepsisBundleMonitor_CompletedBundle_NotMarkedOverdue()
|
||||
{
|
||||
var (_, encounterId) = await SeedPatientEncounterAndBundleAsync();
|
||||
|
||||
using (var scope = _fixture.Services.CreateScope())
|
||||
{
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
var bundle = await db.SepsisBundles.SingleAsync(b => b.EncounterId == encounterId);
|
||||
bundle.ComplianceStatus = SepsisBundleComplianceStatus.Compliant;
|
||||
bundle.CompletedAt = DateTimeOffset.UtcNow;
|
||||
bundle.DeadlineAt = DateTimeOffset.UtcNow.AddHours(-1);
|
||||
await db.SaveChangesAsync();
|
||||
}
|
||||
|
||||
var monitor = new SepsisBundleMonitorService(
|
||||
_fixture.Services,
|
||||
_fixture.Services.GetRequiredService<ClinicalMetrics>(),
|
||||
_fixture.Services.GetRequiredService<ILogger<SepsisBundleMonitorService>>());
|
||||
|
||||
await monitor.ScanOverdueBundlesAsync(CancellationToken.None);
|
||||
|
||||
using var verifyScope = _fixture.Services.CreateScope();
|
||||
var verifyDb = verifyScope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var bundle2 = await verifyDb.SepsisBundles
|
||||
.AsNoTracking()
|
||||
.SingleAsync(b => b.EncounterId == encounterId);
|
||||
|
||||
bundle2.ComplianceStatus.Should().Be(SepsisBundleComplianceStatus.Compliant,
|
||||
"already-completed bundles must not be marked non-compliant by the monitor");
|
||||
}
|
||||
|
||||
// =======================================================================
|
||||
// EscalationWorker — only escalates open alerts
|
||||
// =======================================================================
|
||||
|
||||
[Fact]
|
||||
public async Task EscalationWorker_AcknowledgedAlert_NotEscalated()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var (patientId, encounterId) = await SeedPatientAndEncounterAsync();
|
||||
|
||||
var alertId = Guid.NewGuid();
|
||||
db.ClinicalAlerts.Add(new ClinicalAlert
|
||||
{
|
||||
Id = alertId,
|
||||
EncounterId = encounterId,
|
||||
PatientId = patientId,
|
||||
AlertType = AlertType.SofaSepsis,
|
||||
Severity = AlertSeverity.Critical,
|
||||
Details = "SOFA delta +2",
|
||||
Status = AlertStatus.Acknowledged,
|
||||
TriggeredAt = DateTimeOffset.UtcNow
|
||||
});
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
var escalationService = new EscalationWorkerService(
|
||||
_fixture.Services.GetRequiredService<Microsoft.Extensions.Options.IOptions<RabbitMqOptions>>(),
|
||||
_fixture.Services.GetRequiredService<IServiceScopeFactory>(),
|
||||
_fixture.Services.GetRequiredService<ClinicalMetrics>(),
|
||||
_fixture.Services.GetRequiredService<ILogger<EscalationWorkerService>>());
|
||||
|
||||
var escalated = await InvokeUpdateAlertStatusEscalatedAsync(escalationService, alertId);
|
||||
|
||||
escalated.Should().BeFalse("acknowledged alerts must not be escalated");
|
||||
|
||||
var alert = await db.ClinicalAlerts.AsNoTracking().SingleAsync(a => a.Id == alertId);
|
||||
alert.Status.Should().Be(AlertStatus.Acknowledged);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task EscalationWorker_ResolvedAlert_NotEscalated()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var (patientId, encounterId) = await SeedPatientAndEncounterAsync();
|
||||
|
||||
var alertId = Guid.NewGuid();
|
||||
db.ClinicalAlerts.Add(new ClinicalAlert
|
||||
{
|
||||
Id = alertId,
|
||||
EncounterId = encounterId,
|
||||
PatientId = patientId,
|
||||
AlertType = AlertType.SofaSepsis,
|
||||
Severity = AlertSeverity.Critical,
|
||||
Details = "SOFA delta +2",
|
||||
Status = AlertStatus.Resolved,
|
||||
TriggeredAt = DateTimeOffset.UtcNow
|
||||
});
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
var escalationService = new EscalationWorkerService(
|
||||
_fixture.Services.GetRequiredService<Microsoft.Extensions.Options.IOptions<RabbitMqOptions>>(),
|
||||
_fixture.Services.GetRequiredService<IServiceScopeFactory>(),
|
||||
_fixture.Services.GetRequiredService<ClinicalMetrics>(),
|
||||
_fixture.Services.GetRequiredService<ILogger<EscalationWorkerService>>());
|
||||
|
||||
var escalated = await InvokeUpdateAlertStatusEscalatedAsync(escalationService, alertId);
|
||||
|
||||
escalated.Should().BeFalse("resolved alerts must not be escalated");
|
||||
|
||||
var alert = await db.ClinicalAlerts.AsNoTracking().SingleAsync(a => a.Id == alertId);
|
||||
alert.Status.Should().Be(AlertStatus.Resolved);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task EscalationWorker_OpenAlert_IsEscalated()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var (patientId, encounterId) = await SeedPatientAndEncounterAsync();
|
||||
|
||||
var alertId = Guid.NewGuid();
|
||||
db.ClinicalAlerts.Add(new ClinicalAlert
|
||||
{
|
||||
Id = alertId,
|
||||
EncounterId = encounterId,
|
||||
PatientId = patientId,
|
||||
AlertType = AlertType.SofaSepsis,
|
||||
Severity = AlertSeverity.Critical,
|
||||
Details = "SOFA delta +2",
|
||||
Status = AlertStatus.Open,
|
||||
TriggeredAt = DateTimeOffset.UtcNow
|
||||
});
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
var escalationService = new EscalationWorkerService(
|
||||
_fixture.Services.GetRequiredService<Microsoft.Extensions.Options.IOptions<RabbitMqOptions>>(),
|
||||
_fixture.Services.GetRequiredService<IServiceScopeFactory>(),
|
||||
_fixture.Services.GetRequiredService<ClinicalMetrics>(),
|
||||
_fixture.Services.GetRequiredService<ILogger<EscalationWorkerService>>());
|
||||
|
||||
var escalated = await InvokeUpdateAlertStatusEscalatedAsync(escalationService, alertId);
|
||||
|
||||
escalated.Should().BeTrue("open alerts must be escalated");
|
||||
|
||||
var alert = await db.ClinicalAlerts.AsNoTracking().SingleAsync(a => a.Id == alertId);
|
||||
alert.Status.Should().Be(AlertStatus.Escalated);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task EscalationWorker_NonexistentAlert_ReturnsFalse()
|
||||
{
|
||||
var escalationService = new EscalationWorkerService(
|
||||
_fixture.Services.GetRequiredService<Microsoft.Extensions.Options.IOptions<RabbitMqOptions>>(),
|
||||
_fixture.Services.GetRequiredService<IServiceScopeFactory>(),
|
||||
_fixture.Services.GetRequiredService<ClinicalMetrics>(),
|
||||
_fixture.Services.GetRequiredService<ILogger<EscalationWorkerService>>());
|
||||
|
||||
var escalated = await InvokeUpdateAlertStatusEscalatedAsync(escalationService, Guid.NewGuid());
|
||||
|
||||
escalated.Should().BeFalse("nonexistent alerts must not cause errors");
|
||||
}
|
||||
|
||||
// =======================================================================
|
||||
// Cancellation — background services respect CancellationToken
|
||||
// =======================================================================
|
||||
|
||||
[Fact]
|
||||
public async Task SepsisBundleMonitor_CancellationRequested_StopsGracefully()
|
||||
{
|
||||
var monitor = new SepsisBundleMonitorService(
|
||||
_fixture.Services,
|
||||
_fixture.Services.GetRequiredService<ClinicalMetrics>(),
|
||||
_fixture.Services.GetRequiredService<ILogger<SepsisBundleMonitorService>>());
|
||||
|
||||
using var cts = new CancellationTokenSource();
|
||||
cts.Cancel();
|
||||
|
||||
var act = () => monitor.ScanOverdueBundlesAsync(cts.Token);
|
||||
|
||||
await act.Should().ThrowAsync<OperationCanceledException>();
|
||||
}
|
||||
|
||||
// =======================================================================
|
||||
// Helpers
|
||||
// =======================================================================
|
||||
|
||||
private static ConsumeResult<string, string> BuildConsumeResult(
|
||||
string topic, int partition, long offset, string value)
|
||||
{
|
||||
return new ConsumeResult<string, string>
|
||||
{
|
||||
Topic = topic,
|
||||
Partition = new Partition(partition),
|
||||
Offset = new Offset(offset),
|
||||
Message = new Message<string, string>
|
||||
{
|
||||
Key = "test-key",
|
||||
Value = value
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private async Task<(Guid PatientId, Guid EncounterId)> SeedPatientAndEncounterAsync()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var patient = new Patient
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
Mrn = $"MRN-BG-{Guid.NewGuid():N}"[..20],
|
||||
FirstName = "Background",
|
||||
LastName = "Test",
|
||||
DateOfBirth = new DateOnly(1970, 1, 1),
|
||||
Gender = "M",
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
var encounter = new Encounter
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
PatientId = patient.Id,
|
||||
EncounterType = EncounterType.Inpatient,
|
||||
Status = EncounterStatus.Active,
|
||||
Department = Department.Icu,
|
||||
AttendingPhysician = "Dr. Background",
|
||||
AdmittedAt = DateTimeOffset.UtcNow,
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
|
||||
db.Patients.Add(patient);
|
||||
db.Encounters.Add(encounter);
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
return (patient.Id, encounter.Id);
|
||||
}
|
||||
|
||||
private async Task<(Guid PatientId, Guid EncounterId)> SeedPatientEncounterAndBundleAsync()
|
||||
{
|
||||
var (patientId, encounterId) = await SeedPatientAndEncounterAsync();
|
||||
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
var handler = scope.ServiceProvider.GetRequiredService<SepsisAlertHandler>();
|
||||
|
||||
var alertId = Guid.NewGuid();
|
||||
db.ClinicalAlerts.Add(new ClinicalAlert
|
||||
{
|
||||
Id = alertId,
|
||||
EncounterId = encounterId,
|
||||
PatientId = patientId,
|
||||
AlertType = AlertType.SofaSepsis,
|
||||
Severity = AlertSeverity.Critical,
|
||||
Details = "SOFA delta +2",
|
||||
Status = AlertStatus.Open,
|
||||
TriggeredAt = DateTimeOffset.UtcNow
|
||||
});
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
await handler.OnSepsisAlertCreatedAsync(
|
||||
encounterId, alertId, AlertType.SofaSepsis, CancellationToken.None);
|
||||
|
||||
return (patientId, encounterId);
|
||||
}
|
||||
|
||||
private static async Task<bool> InvokeUpdateAlertStatusEscalatedAsync(
|
||||
EscalationWorkerService service, Guid alertId)
|
||||
{
|
||||
var method = typeof(EscalationWorkerService).GetMethod(
|
||||
"UpdateAlertStatusEscalatedAsync",
|
||||
System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance);
|
||||
|
||||
var task = (Task<bool>)method!.Invoke(service, new object[] { alertId, CancellationToken.None })!;
|
||||
return await task;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,340 @@
|
||||
using System.Net;
|
||||
using System.Net.Http.Json;
|
||||
using System.Text.Json;
|
||||
using FluentAssertions;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using StackExchange.Redis;
|
||||
|
||||
[Collection("Integration")]
|
||||
public class ConcurrencyTests : IAsyncLifetime
|
||||
{
|
||||
private readonly ApiFixture _fixture;
|
||||
private readonly HttpClient _http;
|
||||
|
||||
public ConcurrencyTests(ApiFixture fixture)
|
||||
{
|
||||
_fixture = fixture;
|
||||
_http = fixture.CreateClient();
|
||||
_http.ClearAuth();
|
||||
_http.AsAdmin();
|
||||
}
|
||||
|
||||
public async Task InitializeAsync()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
await DbResetHelper.ResetAsync(db);
|
||||
|
||||
var redis = scope.ServiceProvider.GetRequiredService<IConnectionMultiplexer>();
|
||||
var server = redis.GetServer(redis.GetEndPoints().First());
|
||||
await server.FlushDatabaseAsync(1);
|
||||
}
|
||||
|
||||
public Task DisposeAsync() => Task.CompletedTask;
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Parallel patient registration — each must get a unique MRN
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task ParallelPatientRegistration_AllGetUniqueMrns()
|
||||
{
|
||||
const int parallelCount = 10;
|
||||
|
||||
var tasks = Enumerable.Range(0, parallelCount).Select(i =>
|
||||
{
|
||||
var client = _fixture.CreateClient();
|
||||
client.ClearAuth();
|
||||
client.AsAdmin();
|
||||
return client.PostAsJsonAsync(
|
||||
"/api/v1/patients",
|
||||
new RegisterPatientRequest(
|
||||
$"Parallel{i}", "Patient",
|
||||
new DateOnly(1980, 1, 1).AddDays(i), "Female"));
|
||||
});
|
||||
|
||||
var responses = await Task.WhenAll(tasks);
|
||||
|
||||
var patientIds = new List<Guid>();
|
||||
foreach (var resp in responses)
|
||||
{
|
||||
resp.StatusCode.Should().Be(HttpStatusCode.Created);
|
||||
var body = await resp.Content.ReadFromJsonAsync<JsonDocument>();
|
||||
patientIds.Add(body!.RootElement.GetProperty("data").GetProperty("id").GetGuid());
|
||||
}
|
||||
|
||||
patientIds.Should().OnlyHaveUniqueItems("every parallel registration must receive a distinct ID");
|
||||
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var mrns = await db.Patients
|
||||
.Where(p => patientIds.Contains(p.Id))
|
||||
.Select(p => p.Mrn)
|
||||
.ToListAsync();
|
||||
|
||||
mrns.Should().HaveCount(parallelCount);
|
||||
mrns.Should().OnlyHaveUniqueItems("every parallel registration must receive a distinct MRN");
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Parallel SOFA_SEPSIS alerts for same encounter — only one bundle
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task ParallelSepsisAlerts_SingleBundleCreated()
|
||||
{
|
||||
var (patientId, encounterId) = await SeedPatientAndEncounterAsync();
|
||||
const int parallelCount = 5;
|
||||
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var alertIds = new List<Guid>();
|
||||
for (var i = 0; i < parallelCount; i++)
|
||||
{
|
||||
var alertId = Guid.NewGuid();
|
||||
db.ClinicalAlerts.Add(new ClinicalAlert
|
||||
{
|
||||
Id = alertId,
|
||||
EncounterId = encounterId,
|
||||
PatientId = patientId,
|
||||
AlertType = AlertType.SofaSepsis,
|
||||
Severity = AlertSeverity.Critical,
|
||||
Details = $"SOFA delta +{i + 2}",
|
||||
Status = AlertStatus.Open,
|
||||
TriggeredAt = DateTimeOffset.UtcNow
|
||||
});
|
||||
alertIds.Add(alertId);
|
||||
}
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
var tasks = alertIds.Select(alertId =>
|
||||
{
|
||||
return Task.Run(async () =>
|
||||
{
|
||||
using var innerScope = _fixture.Services.CreateScope();
|
||||
var handler = innerScope.ServiceProvider.GetRequiredService<SepsisAlertHandler>();
|
||||
await handler.OnSepsisAlertCreatedAsync(
|
||||
encounterId, alertId, AlertType.SofaSepsis, CancellationToken.None);
|
||||
});
|
||||
});
|
||||
|
||||
await Task.WhenAll(tasks);
|
||||
|
||||
var bundleCount = await db.SepsisBundles
|
||||
.AsNoTracking()
|
||||
.CountAsync(b => b.EncounterId == encounterId);
|
||||
|
||||
bundleCount.Should().Be(1, "concurrent SOFA_SEPSIS alerts must produce exactly one bundle");
|
||||
|
||||
var orderCount = await db.Orders
|
||||
.AsNoTracking()
|
||||
.CountAsync(o => o.EncounterId == encounterId);
|
||||
|
||||
orderCount.Should().Be(4, "only one bundle's orders (4) should exist");
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Parallel observation ingest with same idempotency key — single record
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task ParallelObservationIngest_SameIdempotencyKey_SingleRecord()
|
||||
{
|
||||
var (_, encounterId) = await SeedPatientAndEncounterAsync();
|
||||
await SeedThresholdsAsync();
|
||||
|
||||
const int parallelCount = 5;
|
||||
var idempotencyKey = $"device-parallel-{Guid.NewGuid():N}";
|
||||
|
||||
var tasks = Enumerable.Range(0, parallelCount).Select(_ =>
|
||||
{
|
||||
var client = _fixture.CreateClient();
|
||||
client.ClearAuth();
|
||||
client.AsAdmin();
|
||||
return client.PostAsJsonAsync(
|
||||
$"/api/v1/encounters/{encounterId}/observations",
|
||||
new BatchIngestRequest(new List<IngestObservationRequest>
|
||||
{
|
||||
new("HEART_RATE", 78, "bpm", ObservationSource.Device,
|
||||
DateTimeOffset.UtcNow, idempotencyKey)
|
||||
}));
|
||||
});
|
||||
|
||||
var responses = await Task.WhenAll(tasks);
|
||||
|
||||
foreach (var resp in responses)
|
||||
resp.StatusCode.Should().Be(HttpStatusCode.Created);
|
||||
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var obsCount = await db.Observations
|
||||
.CountAsync(o => o.EncounterId == encounterId);
|
||||
|
||||
obsCount.Should().Be(1, "parallel ingests with the same idempotency key must produce exactly one record");
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Parallel observation ingest without idempotency key — all succeed
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task ParallelObservationIngest_NoIdempotencyKey_AllCreateRecords()
|
||||
{
|
||||
var (_, encounterId) = await SeedPatientAndEncounterAsync();
|
||||
await SeedThresholdsAsync();
|
||||
|
||||
const int parallelCount = 5;
|
||||
|
||||
var tasks = Enumerable.Range(0, parallelCount).Select(i =>
|
||||
{
|
||||
var client = _fixture.CreateClient();
|
||||
client.ClearAuth();
|
||||
client.AsAdmin();
|
||||
return client.PostAsJsonAsync(
|
||||
$"/api/v1/encounters/{encounterId}/observations",
|
||||
new BatchIngestRequest(new List<IngestObservationRequest>
|
||||
{
|
||||
new("HEART_RATE", 70 + i, "bpm", ObservationSource.Device,
|
||||
DateTimeOffset.UtcNow, null)
|
||||
}));
|
||||
});
|
||||
|
||||
var responses = await Task.WhenAll(tasks);
|
||||
|
||||
foreach (var resp in responses)
|
||||
resp.StatusCode.Should().Be(HttpStatusCode.Created);
|
||||
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var obsCount = await db.Observations
|
||||
.CountAsync(o => o.EncounterId == encounterId);
|
||||
|
||||
obsCount.Should().Be(parallelCount,
|
||||
"each observation without an idempotency key is distinct and must be stored");
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Parallel encounter open — duplicate active encounter rejected
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task ParallelEncounterOpen_SameType_OnlyOneSucceeds()
|
||||
{
|
||||
var patientId = await CreatePatientViaApiAsync();
|
||||
const int parallelCount = 5;
|
||||
|
||||
var tasks = Enumerable.Range(0, parallelCount).Select(_ =>
|
||||
{
|
||||
var client = _fixture.CreateClient();
|
||||
client.ClearAuth();
|
||||
client.AsAdmin();
|
||||
return client.PostAsJsonAsync(
|
||||
$"/api/v1/patients/{patientId}/encounters",
|
||||
new OpenEncounterRequest(EncounterType.Inpatient, Department.Icu, "Dr. Race"));
|
||||
});
|
||||
|
||||
var responses = await Task.WhenAll(tasks);
|
||||
|
||||
var created = responses.Count(r => r.StatusCode == HttpStatusCode.Created);
|
||||
var conflicts = responses.Count(r => r.StatusCode == HttpStatusCode.Conflict);
|
||||
|
||||
created.Should().BeGreaterThanOrEqualTo(1, "at least one encounter must be created");
|
||||
(created + conflicts).Should().Be(parallelCount,
|
||||
"all responses must be either Created or Conflict");
|
||||
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var activeCount = await db.Encounters
|
||||
.CountAsync(e => e.PatientId == patientId
|
||||
&& e.EncounterType == EncounterType.Inpatient
|
||||
&& e.Status == EncounterStatus.Active);
|
||||
|
||||
activeCount.Should().Be(1,
|
||||
"only one active encounter of the same type should exist per patient");
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Helpers
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
private async Task<(Guid PatientId, Guid EncounterId)> SeedPatientAndEncounterAsync()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var patient = new Patient
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
Mrn = $"MRN-CONC-{Guid.NewGuid():N}"[..20],
|
||||
FirstName = "Concurrent",
|
||||
LastName = "Test",
|
||||
DateOfBirth = new DateOnly(1975, 6, 15),
|
||||
Gender = "M",
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
var encounter = new Encounter
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
PatientId = patient.Id,
|
||||
EncounterType = EncounterType.Inpatient,
|
||||
Status = EncounterStatus.Active,
|
||||
Department = Department.Icu,
|
||||
AttendingPhysician = "Dr. Concurrent",
|
||||
AdmittedAt = DateTimeOffset.UtcNow,
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
|
||||
db.Patients.Add(patient);
|
||||
db.Encounters.Add(encounter);
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
return (patient.Id, encounter.Id);
|
||||
}
|
||||
|
||||
private async Task SeedThresholdsAsync()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
if (!await db.AlertThresholds.AnyAsync(t => t.ObservationCode == "HEART_RATE"))
|
||||
{
|
||||
db.AlertThresholds.Add(new AlertThreshold
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
ObservationCode = "HEART_RATE",
|
||||
DisplayName = "Heart Rate",
|
||||
Unit = "bpm",
|
||||
CriticalLow = 30,
|
||||
WarningLow = 50,
|
||||
WarningHigh = 100,
|
||||
CriticalHigh = 150,
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
});
|
||||
await db.SaveChangesAsync();
|
||||
}
|
||||
|
||||
var redis = scope.ServiceProvider.GetRequiredService<IConnectionMultiplexer>();
|
||||
await redis.GetDatabase(1).StringSetAsync(
|
||||
"threshold:HEART_RATE",
|
||||
"""{"ObservationCode":"HEART_RATE","CriticalLow":30,"WarningLow":50,"WarningHigh":100,"CriticalHigh":150}""");
|
||||
}
|
||||
|
||||
private async Task<Guid> CreatePatientViaApiAsync()
|
||||
{
|
||||
var resp = await _http.PostAsJsonAsync(
|
||||
"/api/v1/patients",
|
||||
new RegisterPatientRequest("Race", "Condition",
|
||||
new DateOnly(1985, 3, 20), "Male"));
|
||||
resp.EnsureSuccessStatusCode();
|
||||
|
||||
var body = await resp.Content.ReadFromJsonAsync<JsonDocument>();
|
||||
return body!.RootElement.GetProperty("data").GetProperty("id").GetGuid();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user