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

DotNet/Normalization/Application/Error/ResourcesAcquiredRetryDeadLetterHandler.cs

@@ -0,0 +1,89 @@
+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 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);
+            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,91 @@
+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>
+/// Unlike the success path — which deletes only the per-resource-type acquisition keys and leaves
+/// <c>{correlationId}</c> in place for Measure Eval to read — a terminal failure also removes the
+/// correlation key, because nothing downstream will ever consume it.
+/// </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;
+        }
+
+        // Acquisition keys are "{correlationId}:{ResourceType}"; the correlation key itself holds
+        // whatever normalization managed to write before it failed, so it goes too.
+        var keysToDelete = new List<string>(cacheKeys);
+        keysToDelete.AddRange(cacheKeys
+            .Select(ExtractCorrelationId)
+            .Where(correlationId => !string.IsNullOrWhiteSpace(correlationId))
+            .Distinct()
+            .Where(correlationId => !keysToDelete.Contains(correlationId)));
+
+        try
+        {
+            await _resourceCache
+                .GetImplementation(value!.CacheType)
+                .DeleteAsync(keysToDelete, cancellationToken);
+
+            _logger.LogInformation(
+                "Purged {KeyCount} resource cache entries after terminal failure ({Reason}). CacheType: {CacheType}, Keys: [{Keys}]",
+                keysToDelete.Count,
+                reason.SanitizeForLog(),
+                value.CacheType,
+                string.Join(", ", keysToDelete).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(", ", keysToDelete).SanitizeForLog());
+        }
+    }
+
+    /// <remarks>Mirrors <c>HybridResourceCache.ExtractCorrelationId</c>.</remarks>
+    private static string ExtractCorrelationId(string cacheKey)
+    {
+        var idx = cacheKey.IndexOf(':');
+        return idx > 0 ? cacheKey[..idx] : cacheKey;
+    }
+}

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,47 @@ 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 cached resources. The retry paths below must NOT do this — a redelivered
+            // message still needs its cache.
+            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,13 @@ static void RegisterServices(WebApplicationBuilder builder)
     builder.Services.AddSingleton(typeof(ITransientExceptionHandler<,,>), typeof(TransientExceptionHandler<,,>));
     builder.Services.AddSingleton(typeof(IDeadLetterExceptionHandler<,,>), typeof(DeadLetterExceptionHandler<,,>));
 
+    builder.Services.AddSingleton<IResourceCachePurger, ResourceCachePurger>();
+
+    // Registered after the open generic above so that it wins for this closed type: 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()