From a91d79f3cd6f1e0593f222b7e5e2c4adbb825898 Mon Sep 17 00:00:00 2001 From: voltsrage Date: Sun, 21 Jun 2026 17:01:16 +0800 Subject: [PATCH] Fix: FHIR bundle processing has no rollback on partial failure --- .../Fhir/FhirBundleProcessor.cs | 12 ++++++++++-- .../Services/ObservationService.cs | 18 +++++++----------- VigilCareClinicalAPI/Services/OrderService.cs | 5 +++++ 3 files changed, 22 insertions(+), 13 deletions(-) diff --git a/VigilCareClinicalAPI/Fhir/FhirBundleProcessor.cs b/VigilCareClinicalAPI/Fhir/FhirBundleProcessor.cs index 21853ee..0719118 100644 --- a/VigilCareClinicalAPI/Fhir/FhirBundleProcessor.cs +++ b/VigilCareClinicalAPI/Fhir/FhirBundleProcessor.cs @@ -1,7 +1,9 @@ using Hl7.Fhir.Model; +using Microsoft.EntityFrameworkCore; public class FhirBundleProcessor { + private readonly AppDbContext _db; private readonly IPatientService _patients; private readonly IEncounterService _encounters; private readonly IObservationService _observations; @@ -12,6 +14,7 @@ public class FhirBundleProcessor private readonly MedicationAdministrationFhirMapper _medMapper; public FhirBundleProcessor( + AppDbContext db, IPatientService patients, IEncounterService encounters, IObservationService observations, @@ -21,6 +24,7 @@ public class FhirBundleProcessor ObservationFhirMapper observationMapper, MedicationAdministrationFhirMapper medMapper) { + _db = db; _patients = patients; _encounters = encounters; _observations = observations; @@ -35,11 +39,12 @@ public class FhirBundleProcessor { var response = new Bundle { Type = Bundle.BundleType.TransactionResponse }; - // Process in dependency order: Patient → Encounter → Observation/MedAdmin var entries = transaction.Entry .OrderBy(e => Priority(e.Resource)) .ToList(); + await using var tx = await _db.Database.BeginTransactionAsync(); + foreach (var entry in entries) { var resource = entry.Resource; @@ -66,6 +71,8 @@ public class FhirBundleProcessor } catch (Exception ex) { + await tx.RollbackAsync(); + response.Entry.Add(new Bundle.EntryComponent { Response = new Bundle.ResponseComponent @@ -74,10 +81,11 @@ public class FhirBundleProcessor Outcome = FhirOperationOutcomeBuilder.FromException(ex) } }); - break; // transaction semantics — stop on first failure + return response; } } + await tx.CommitAsync(); return response; } diff --git a/VigilCareClinicalAPI/Services/ObservationService.cs b/VigilCareClinicalAPI/Services/ObservationService.cs index 2641dbb..9437a80 100644 --- a/VigilCareClinicalAPI/Services/ObservationService.cs +++ b/VigilCareClinicalAPI/Services/ObservationService.cs @@ -66,7 +66,10 @@ public class ObservationService : IObservationService { using var timer = _metrics.ObservationIngestDuration.NewTimer(); - await using var tx = await _db.Database.BeginTransactionAsync(); + // Re-use an outer transaction (e.g. FhirBundleProcessor) when one is already active. + await using var tx = _db.Database.CurrentTransaction is null + ? await _db.Database.BeginTransactionAsync() + : null; try { // Step 4 — insert observation @@ -96,10 +99,6 @@ public class ObservationService : IObservationService ClinicalAlert? alert = null; // Step 6 — critical threshold detection (synchronous) - // WARNING detection is intentionally deferred to the Kafka consumer. - // A critical potassium of 2.1 mEq/L is immediately life-threatening — the alert - // must exist before this API call returns. A warning heart rate of 95 bpm warrants - // attention but not an emergency page; the additional Kafka latency is clinically safe. if (IsCriticalBreach(req.Value, threshold)) { alert = new ClinicalAlert @@ -116,7 +115,6 @@ public class ObservationService : IObservationService }; _db.ClinicalAlerts.Add(alert); - // Step 6b — outbox event for the alert (relay picks this up in Phase 3) _db.OutboxEvents.Add(BuildOutboxEvent("alert.generated", new { alertId = alert.Id, @@ -149,7 +147,7 @@ public class ObservationService : IObservationService // Step 8 — COMMIT await _db.SaveChangesAsync(); - await tx.CommitAsync(); + if (tx is not null) await tx.CommitAsync(); _metrics.ObservationsIngestedTotal .WithLabels(req.ObservationCode, req.Source.ToDbString()) @@ -173,9 +171,7 @@ public class ObservationService : IObservationService } catch (DbUpdateException ex) when (IsUniqueViolation(ex)) { - // Race condition: two concurrent retries both passed the pre-check above. - // The unique partial index caught it. Roll back and return the existing row. - await tx.RollbackAsync(); + if (tx is not null) await tx.RollbackAsync(); var existing = await _db.Observations .AsNoTracking() .FirstOrDefaultAsync(o => o.IdempotencyKey == req.IdempotencyKey); @@ -185,7 +181,7 @@ public class ObservationService : IObservationService } catch { - await tx.RollbackAsync(); + if (tx is not null) await tx.RollbackAsync(); throw; } } diff --git a/VigilCareClinicalAPI/Services/OrderService.cs b/VigilCareClinicalAPI/Services/OrderService.cs index ba4f9d7..3990955 100644 --- a/VigilCareClinicalAPI/Services/OrderService.cs +++ b/VigilCareClinicalAPI/Services/OrderService.cs @@ -113,9 +113,14 @@ public class OrderService : IOrderService order.ResultedAt = DateTimeOffset.UtcNow; order.ResultSummary = req.ResultSummary; + await using var tx = _db.Database.CurrentTransaction is null + ? await _db.Database.BeginTransactionAsync() + : null; await _db.SaveChangesAsync(); var bundleService = _serviceProvider.GetRequiredService(); await bundleService.OnOrderResultedAsync(order.Id); + if (tx is not null) await tx.CommitAsync(); + return order; } } \ No newline at end of file