LEGLINK-958: Measure Eval: clean up the resource cache when evaluation or normalization fails

.github/workflows/tests.yaml

@@ -297,12 +297,23 @@ jobs:
             ${{ runner.os }}-nuget-
 
       # Pulled up-front so the integration fixtures (Testcontainers) don't
-      # spend their per-test pull budget on the cold pull.
+      # spend their per-test pull budget on the cold pull. Retry up to 3 times
+      # to handle transient MCR network timeouts.
       - name: Pull Azurite Docker image
-        run: docker pull mcr.microsoft.com/azure-storage/azurite:latest
+        run: |
+          for i in 1 2 3; do
+            docker pull mcr.microsoft.com/azure-storage/azurite:latest && break
+            echo "Pull attempt $i failed, retrying in 15s..."
+            sleep 15
+          done
 
       - name: Pull SQL Server Docker image
-        run: docker pull mcr.microsoft.com/mssql/server:2022-CU13-ubuntu-22.04
+        run: |
+          for i in 1 2 3; do
+            docker pull mcr.microsoft.com/mssql/server:2022-CU13-ubuntu-22.04 && break
+            echo "Pull attempt $i failed, retrying in 15s..."
+            sleep 15
+          done
 
       - name: Restore dependencies
         run: dotnet restore DotNet/ServiceTests/ServiceTests.csproj

DotNet/Normalization/Application/Error/ResourcesAcquiredRetryDeadLetterHandler.cs

@@ -0,0 +1,90 @@
+using Confluent.Kafka;
+using LantanaGroup.Link.Normalization.Application.Models.Messages;
+using LantanaGroup.Link.Normalization.Application.Services;
+using LantanaGroup.Link.Shared.Application.Error.Handlers;
+using LantanaGroup.Link.Shared.Application.Error.Interfaces;
+using LantanaGroup.Link.Shared.Application.Interfaces;
+using LantanaGroup.Link.Shared.Application.Listeners;
+using LantanaGroup.Link.Shared.Application.Models;
+using LantanaGroup.Link.Shared.Application.Services.Security;
+using System.Text.Json;
+
+namespace LantanaGroup.Link.Normalization.Application.Error;
+
+/// <summary>
+/// Dead letter handler for <see cref="RetryListener"/> that also releases the resource cache.
+/// </summary>
+/// <remarks>
+/// A <c>ResourcesAcquired</c> message that fails transiently is republished to
+/// <c>ResourcesAcquired-Retry</c> and redelivered on a schedule. Once the retry count is exhausted,
+/// <see cref="RetryListener"/> — shared, service-agnostic code with no knowledge of the resource
+/// cache — dead-letters it to <c>ResourcesAcquired-Error</c>. That is the second and final terminal
+/// path for the message (the first being a <c>DeadLetterException</c> raised directly in
+/// <c>ResourcesAcquiredListener</c>), so it is where the cached resources have to be released.
+/// <para>
+/// Normalization registers <see cref="RetryListener"/> for <c>ResourcesAcquired-Retry</c> only, so
+/// every message reaching this handler is a <see cref="ResourcesAcquiredValue"/>.
+/// </para>
+/// </remarks>
+public class ResourcesAcquiredRetryDeadLetterHandler : DeadLetterExceptionHandler<RetryListener, string, string>
+{
+    private static readonly JsonSerializerOptions DeserializerOptions = new()
+    {
+        // Mirrors JsonWithFhirMessageDeserializer, which is how the value was read off the topic.
+        PropertyNameCaseInsensitive = true,
+        AllowTrailingCommas = true
+    };
+
+    private readonly IResourceCachePurger _resourceCachePurger;
+    private readonly ILogger<ResourcesAcquiredRetryDeadLetterHandler> _logger;
+
+    public ResourcesAcquiredRetryDeadLetterHandler(
+        IKafkaProducerFactory<string, string> producerFactory,
+        IKafkaProducerFactory<string, string> nullConsumeResultProducerFactory,
+        ServiceInformation serviceInformation,
+        IExceptionLogger<RetryListener> exceptionHandler,
+        IResourceCachePurger resourceCachePurger,
+        ILogger<ResourcesAcquiredRetryDeadLetterHandler> logger)
+        : base(producerFactory, nullConsumeResultProducerFactory, serviceInformation, exceptionHandler)
+    {
+        _resourceCachePurger = resourceCachePurger ?? throw new ArgumentNullException(nameof(resourceCachePurger));
+        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+    }
+
+    /// <remarks>
+    /// Both <c>HandleException</c> overloads funnel through here. The dead letter is produced first so
+    /// that the durable record of the failure is never at the mercy of the cache purge.
+    /// </remarks>
+    public override void ProduceDeadLetter(ConsumeResult<string, string> consumeResult, string exceptionMessage)
+    {
+        base.ProduceDeadLetter(consumeResult, exceptionMessage);
+
+        PurgeResourceCache(consumeResult, exceptionMessage);
+    }
+
+    private void PurgeResourceCache(ConsumeResult<string, string> consumeResult, string exceptionMessage)
+    {
+        ResourcesAcquiredValue? value;
+
+        try
+        {
+            value = JsonSerializer.Deserialize<ResourcesAcquiredValue>(
+                consumeResult.Message.Value, DeserializerOptions);
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex,
+                "Could not deserialize a retry-exhausted message from {Topic} to determine its resource cache keys. " +
+                "Any cached resources for it will be released by the cache expiration policy instead.",
+                consumeResult.Topic.SanitizeForLog());
+            return;
+        }
+
+        // RetryListener runs its consume callback on a background thread with no synchronization
+        // context, so blocking here cannot deadlock. PurgeAsync does not throw.
+        _resourceCachePurger
+            .PurgeAsync(value, $"retry count exhausted: {exceptionMessage}")
+            .GetAwaiter()
+            .GetResult();
+    }
+}

DotNet/Normalization/Application/Services/ResourceCachePurger.cs

@@ -0,0 +1,79 @@
+using LantanaGroup.Link.Normalization.Application.Models.Messages;
+using LantanaGroup.Link.Shared.Application.Interfaces;
+using LantanaGroup.Link.Shared.Application.Services.Security;
+
+namespace LantanaGroup.Link.Normalization.Application.Services;
+
+/// <summary>
+/// Releases the resource cache entries belonging to a <c>ResourcesAcquired</c> message after a
+/// terminal (non-retryable) normalization failure.
+/// </summary>
+/// <remarks>
+/// Only call this on terminal failures. A message bound for <c>ResourcesAcquired-Retry</c> still
+/// needs its cached resources when it is redelivered, so purging on a transient failure would
+/// guarantee the retry fails too.
+/// <para>
+/// Only the per-resource-type acquisition keys (<c>{correlationId}:{ResourceType}</c>) are ever
+/// deleted — they are Normalization's input, consumed by this message alone. The
+/// <c>{correlationId}</c> key is Normalization's output and belongs to its reader: Measure Eval
+/// deletes it after evaluation, and the cache expiration policy reclaims it if no reader ever
+/// comes (e.g. the failure occurred before <c>ResourcesNormalized</c> was published). Normalization
+/// never deletes it, on any path — it cannot know whether an earlier attempt already published, or
+/// whether the key still holds kept INITIAL-phase data a SUPPLEMENTAL evaluation needs.
+/// </para>
+/// </remarks>
+public interface IResourceCachePurger
+{
+    Task PurgeAsync(ResourcesAcquiredValue? value, string reason, CancellationToken cancellationToken = default);
+}
+
+/// <inheritdoc cref="IResourceCachePurger"/>
+public class ResourceCachePurger : IResourceCachePurger
+{
+    private readonly IResourceCache _resourceCache;
+    private readonly ILogger<ResourceCachePurger> _logger;
+
+    public ResourceCachePurger(IResourceCache resourceCache, ILogger<ResourceCachePurger> logger)
+    {
+        _resourceCache = resourceCache ?? throw new ArgumentNullException(nameof(resourceCache));
+        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+    }
+
+    public async Task PurgeAsync(ResourcesAcquiredValue? value, string reason, CancellationToken cancellationToken = default)
+    {
+        var cacheKeys = value?.CacheKeys?.Where(k => !string.IsNullOrWhiteSpace(k)).ToList();
+
+        if (cacheKeys == null || cacheKeys.Count == 0)
+        {
+            _logger.LogWarning(
+                "Cannot purge resource cache after terminal failure ({Reason}): the message carries no cache keys. " +
+                "Any cached resources for it will be released by the cache expiration policy instead.",
+                reason.SanitizeForLog());
+            return;
+        }
+
+        try
+        {
+            await _resourceCache
+                .GetImplementation(value!.CacheType)
+                .DeleteAsync(cacheKeys, cancellationToken);
+
+            _logger.LogInformation(
+                "Purged {KeyCount} resource cache entries after terminal failure ({Reason}). CacheType: {CacheType}, Keys: [{Keys}]",
+                cacheKeys.Count,
+                reason.SanitizeForLog(),
+                value.CacheType,
+                string.Join(", ", cacheKeys).SanitizeForLog());
+        }
+        catch (Exception ex)
+        {
+            // Never let cleanup failure escape: the caller is already handling a failed message, and
+            // the cache expiration policy is the backstop for whatever we could not delete here.
+            _logger.LogError(ex,
+                "Failed to purge resource cache after terminal failure ({Reason}). CacheType: {CacheType}, Keys: [{Keys}]",
+                reason.SanitizeForLog(),
+                value!.CacheType,
+                string.Join(", ", cacheKeys).SanitizeForLog());
+        }
+    }
+}

DotNet/Normalization/Listeners/ResourcesAcquiredListener.cs

@@ -44,6 +44,7 @@ public class ResourcesAcquiredListener : BackgroundService
     private readonly CopyLocationAliasToTypeIterativelyOperationService _copyLocationAliasToTypeIterativelyOperationService;
     private readonly RemoveExtensionsOperationService _removeExtensionsOperationService;
     private readonly IResourceCache _resourceCache;
+    private readonly IResourceCachePurger _resourceCachePurger;
 
     public ResourcesAcquiredListener(
         ILogger<ResourcesAcquiredListener> logger,
@@ -61,7 +62,8 @@ public ResourcesAcquiredListener(
         CopyLocationOperationService copyLocationOperationService,
         CopyLocationAliasToTypeIterativelyOperationService copyLocationAliasToTypeIterativelyOperationService,
         RemoveExtensionsOperationService removeExtensionsOperationService,
-        IResourceCache resourceCache)
+        IResourceCache resourceCache,
+        IResourceCachePurger resourceCachePurger)
     {
         this._logger = logger ?? throw new ArgumentNullException(nameof(logger));
         _consumerFactory = consumerFactory ?? throw new ArgumentNullException(nameof(consumerFactory));
@@ -88,6 +90,7 @@ public ResourcesAcquiredListener(
         _copyLocationAliasToTypeIterativelyOperationService = copyLocationAliasToTypeIterativelyOperationService ?? throw new ArgumentNullException(nameof(copyLocationAliasToTypeIterativelyOperationService));
         _removeExtensionsOperationService = removeExtensionsOperationService ?? throw new ArgumentNullException(nameof(removeExtensionsOperationService));
         _resourceCache = resourceCache ?? throw new ArgumentNullException(nameof(resourceCache));
+        _resourceCachePurger = resourceCachePurger ?? throw new ArgumentNullException(nameof(resourceCachePurger));
     }
 
     protected override async Task ExecuteAsync(CancellationToken cancellationToken)
@@ -113,25 +116,7 @@ await kafkaConsumer.ConsumeWithInstrumentation(async (result, consumeCancellatio
                 {
                     try
                     {
-                        await ProcessMessageAsync(result, consumeCancellationToken);
-                    }
-                    catch (DeadLetterException ex)
-                    {
-                        _deadLetterExceptionHandler.HandleException(result, ex, result.Message.Key?.FacilityId ?? string.Empty);
-                    }
-                    catch (TransientException ex)
-                    {
-                        _transientExceptionHandler.HandleException(result, ex, result.Message.Key?.FacilityId ?? string.Empty);
-                    }
-                    catch (OperationCanceledException) when (consumeCancellationToken.IsCancellationRequested)
-                    {
-                        throw;
-                    }
-                    catch (Exception ex)
-                    {
-                        _logger.LogError(ex, "Failed to process ResourceAcquired event for facility {FacilityId}.", result?.Message.Key?.FacilityId?.SanitizeForLog());
-
-                        _transientExceptionHandler.HandleException(result, new TransientException("Normalization Exception thrown: " + ex.Message, ex), result.Message.Key?.FacilityId ?? string.Empty);
+                        await ConsumeMessageAsync(result, consumeCancellationToken);
                     }
                     finally
                     {
@@ -182,6 +167,48 @@ await kafkaConsumer.ConsumeWithInstrumentation(async (result, consumeCancellatio
         }
     }
 
+    /// <summary>
+    /// Processes a single consumed message and routes any failure to the dead letter or retry topic.
+    /// </summary>
+    /// <remarks>
+    /// Separate from the consume loop (which owns only the offset commit) so that the failure routing —
+    /// in particular which failures release the resource cache — is directly testable.
+    /// </remarks>
+    public async Task ConsumeMessageAsync(ConsumeResult<ResourceKey, ResourcesAcquiredValue> result, CancellationToken consumeCancellationToken)
+    {
+        try
+        {
+            await ProcessMessageAsync(result, consumeCancellationToken);
+        }
+        catch (DeadLetterException ex)
+        {
+            _deadLetterExceptionHandler.HandleException(result, ex, result.Message.Key?.FacilityId ?? string.Empty);
+
+            // Terminal failure: the message is on ResourcesAcquired-Error and will never be normalized,
+            // so release its acquisition keys. The retry paths below must NOT do this — a redelivered
+            // message still needs its cache. The {correlationId} key is never deleted here: Measure
+            // Eval owns its deletion, and the cache expiration policy reclaims it if unread.
+            await _resourceCachePurger.PurgeAsync(
+                result.Message.Value,
+                $"{nameof(KafkaTopic.ResourcesAcquired)} dead-lettered: {ex.Message}",
+                consumeCancellationToken);
+        }
+        catch (TransientException ex)
+        {
+            _transientExceptionHandler.HandleException(result, ex, result.Message.Key?.FacilityId ?? string.Empty);
+        }
+        catch (OperationCanceledException) when (consumeCancellationToken.IsCancellationRequested)
+        {
+            throw;
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex, "Failed to process ResourceAcquired event for facility {FacilityId}.", result?.Message.Key?.FacilityId?.SanitizeForLog());
+
+            _transientExceptionHandler.HandleException(result, new TransientException("Normalization Exception thrown: " + ex.Message, ex), result.Message.Key?.FacilityId ?? string.Empty);
+        }
+    }
+
     public async Task ProcessMessageAsync(ConsumeResult<ResourceKey, ResourcesAcquiredValue> result, CancellationToken cancellationToken)
     {
         ValidateResourcesAcquiredEvent(result, out string correlationId);

DotNet/Normalization/Program.cs

@@ -3,6 +3,7 @@
 using HealthChecks.UI.Client;
 using Hl7.Fhir.Model.CdsHooks;
 using LantanaGroup.Link.Shared.Application.Models.Configs;
+using LantanaGroup.Link.Normalization.Application.Error;
 using LantanaGroup.Link.Normalization.Application.Models.Messages;
 using LantanaGroup.Link.Normalization.Application.Services;
 using LantanaGroup.Link.Normalization.Application.Services.Operations;
@@ -97,6 +98,14 @@ static void RegisterServices(WebApplicationBuilder builder)
     builder.Services.AddSingleton(typeof(ITransientExceptionHandler<,,>), typeof(TransientExceptionHandler<,,>));
     builder.Services.AddSingleton(typeof(IDeadLetterExceptionHandler<,,>), typeof(DeadLetterExceptionHandler<,,>));
 
+    builder.Services.AddSingleton<IResourceCachePurger, ResourceCachePurger>();
+
+    // A closed-type registration takes precedence over the open generic above regardless of the order
+    // the two appear in, so this wins for IDeadLetterExceptionHandler<RetryListener, string, string>:
+    // when RetryListener exhausts the retry count for a ResourcesAcquired message, the dead letter also
+    // releases the resource cache. See ResourcesAcquiredRetryDeadLetterHandler.
+    builder.Services.AddSingleton<IDeadLetterExceptionHandler<RetryListener, string, string>, ResourcesAcquiredRetryDeadLetterHandler>();
+
     builder.Services.AddTransient<ITenantApiService, TenantApiService>();
 
     builder.Services.AddControllers()

DotNet/ServiceTests/IntegrationTests/Normalization/NormalizationIntegrationTestFixture.cs

@@ -159,6 +159,7 @@ public async Task InitializeAsync()
             builder.Services.AddKeyedSingleton<IResourceCache, RedisResourceCache>(ResourceCacheType.Redis);
             builder.Services.AddKeyedSingleton<IResourceCache, ABSResourceCache>(ResourceCacheType.ABS);
             builder.Services.AddSingleton<IResourceCache, HybridResourceCache>();
+            builder.Services.AddSingleton<IResourceCachePurger, ResourceCachePurger>();
 
             builder.Services.AddMemoryCache();