LEGLINK-958: Scope the retry-exhausted purge to acquisition keys only

The retry-exhausted purge deleted the {correlationId} key alongside the
{correlationId}:{Type} acquisition keys, on the premise that "nothing
downstream will ever consume it." That premise does not hold on this path.

ProcessMessageAsync produces ResourcesNormalized BEFORE deleting the
acquisition keys, so a failure at or after the produce - a client-side
produce timeout the broker actually accepted, or the trailing DeleteAsync
throwing - is classed transient and retried while the message is already
out. Measure Eval consumes it, finds the patient reportable, and
deliberately keeps {correlationId} for its SUPPLEMENTAL pass. When the
retries exhaust minutes later, purging that key makes the supplemental
read return empty, and Measure Eval "evaluat[es] with empty bundle to
produce a not-reportable report" - a silent false answer, not an error.

The handler cannot know whether an earlier attempt published; that
knowledge died with the attempt, and no transaction spans Kafka and the
cache. So the purge is now scoped by what each caller CAN prove:

  ResourceCachePurgeScope.All
    Immediate dead-letter path only. DeadLetterException is raised
    exclusively by ValidateResourcesAcquiredEvent, before the processing
    loop, so ResourcesNormalized was provably never produced and the
    correlation key is safe to remove. A comment at the call site pins
    that invariant: a DeadLetterException thrown after the produce would
    invalidate it.

  ResourceCachePurgeScope.AcquisitionKeysOnly
    Retry exhaustion. Only the keys the message itself carries are
    deleted; the derived correlation key is left to the cache expiration
    policy. Cost: one leaked correlation hash per exhausted retry chain.
    The alternative was a silently wrong clinical report.

The scope parameter has no default, so every future call site is forced
to answer "can the publish already have happened?" explicitly.

Tests: new purger test asserting AcquisitionKeysOnly leaves the
correlation key (watched failing with the scope deliberately ignored,
proving it detects the old behavior); both call sites' scope choices
pinned in their existing tests. Normalization unit tests 40/40.

Claude-Session: https://claude.ai/code/session_01AZubeEYCKfAmDCcgVTr18G

Author: arianamihailescu

DotNet/Normalization/Application/Error/ResourcesAcquiredRetryDeadLetterHandler.cs

@@ -80,10 +80,16 @@ private void PurgeResourceCache(ConsumeResult<string, string> consumeResult, str
             return;
         }
 
+        // Retry exhaustion cannot rule out that an earlier attempt already produced
+        // ResourcesNormalized and then failed on the trailing cache delete — the produce precedes
+        // the delete on the success path — in which case Measure Eval is holding {correlationId}
+        // for its SUPPLEMENTAL pass. Deleting it would make that pass read an empty cache and emit
+        // a silent false not-reportable report, so only the acquisition keys are released here.
+        //
         // 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}")
+            .PurgeAsync(value, $"retry count exhausted: {exceptionMessage}", ResourceCachePurgeScope.AcquisitionKeysOnly)
             .GetAwaiter()
             .GetResult();
     }

DotNet/Normalization/Application/Services/ResourceCachePurger.cs

@@ -4,6 +4,30 @@
 
 namespace LantanaGroup.Link.Normalization.Application.Services;
 
+/// <summary>
+/// Which cache keys a terminal-failure purge is allowed to remove. The scope encodes what the
+/// caller can prove about whether <c>ResourcesNormalized</c> was already published for the message.
+/// </summary>
+public enum ResourceCachePurgeScope
+{
+    /// <summary>
+    /// Remove the per-resource-type acquisition keys and the <c>{correlationId}</c> key. Only valid
+    /// where the failure provably occurred before <c>ResourcesNormalized</c> was produced (the
+    /// immediate dead-letter path: validation raises before the processing loop), so nothing
+    /// downstream can be holding the correlation key.
+    /// </summary>
+    All,
+
+    /// <summary>
+    /// Remove only the per-resource-type acquisition keys. For paths that cannot rule out a prior
+    /// publish — retry exhaustion follows attempts that may have produced <c>ResourcesNormalized</c>
+    /// and then failed on the trailing cache delete — where Measure Eval may already be holding
+    /// <c>{correlationId}</c> for its SUPPLEMENTAL pass. The correlation key is left to the cache
+    /// expiration policy.
+    /// </summary>
+    AcquisitionKeysOnly
+}
+
 /// <summary>
 /// Releases the resource cache entries belonging to a <c>ResourcesAcquired</c> message after a
 /// terminal (non-retryable) normalization failure.
@@ -13,14 +37,15 @@ namespace LantanaGroup.Link.Normalization.Application.Services;
 /// 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.
+/// The success path deletes only the per-resource-type acquisition keys and leaves
+/// <c>{correlationId}</c> in place for Measure Eval to read. Whether a terminal failure may also
+/// remove the correlation key depends on what the caller can prove — see
+/// <see cref="ResourceCachePurgeScope"/>.
 /// </para>
 /// </remarks>
 public interface IResourceCachePurger
 {
-    Task PurgeAsync(ResourcesAcquiredValue? value, string reason, CancellationToken cancellationToken = default);
+    Task PurgeAsync(ResourcesAcquiredValue? value, string reason, ResourceCachePurgeScope scope, CancellationToken cancellationToken = default);
 }
 
 /// <inheritdoc cref="IResourceCachePurger"/>
@@ -35,7 +60,7 @@ public ResourceCachePurger(IResourceCache resourceCache, ILogger<ResourceCachePu
         _logger = logger ?? throw new ArgumentNullException(nameof(logger));
     }
 
-    public async Task PurgeAsync(ResourcesAcquiredValue? value, string reason, CancellationToken cancellationToken = default)
+    public async Task PurgeAsync(ResourcesAcquiredValue? value, string reason, ResourceCachePurgeScope scope, CancellationToken cancellationToken = default)
     {
         var cacheKeys = value?.CacheKeys?.Where(k => !string.IsNullOrWhiteSpace(k)).ToList();
 
@@ -48,14 +73,18 @@ public async Task PurgeAsync(ResourcesAcquiredValue? value, string reason, Cance
             return;
         }
 
-        // Acquisition keys are "{correlationId}:{ResourceType}"; the correlation key itself holds
-        // whatever normalization managed to write before it failed, so it goes too.
+        // Acquisition keys are "{correlationId}:{ResourceType}". The correlation key itself is only
+        // removed when the caller can prove ResourcesNormalized was never published for this message
+        // (scope All); otherwise Measure Eval may still be reading it, and it is left to expire.
         var keysToDelete = new List<string>(cacheKeys);
-        keysToDelete.AddRange(cacheKeys
-            .Select(ExtractCorrelationId)
-            .Where(correlationId => !string.IsNullOrWhiteSpace(correlationId))
-            .Distinct()
-            .Where(correlationId => !keysToDelete.Contains(correlationId)));
+        if (scope == ResourceCachePurgeScope.All)
+        {
+            keysToDelete.AddRange(cacheKeys
+                .Select(ExtractCorrelationId)
+                .Where(correlationId => !string.IsNullOrWhiteSpace(correlationId))
+                .Distinct()
+                .Where(correlationId => !keysToDelete.Contains(correlationId)));
+        }
 
         try
         {

DotNet/Normalization/Listeners/ResourcesAcquiredListener.cs

@@ -187,9 +187,16 @@ public async Task ConsumeMessageAsync(ConsumeResult<ResourceKey, ResourcesAcquir
             // 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.
+            //
+            // Scope.All is safe here ONLY because DeadLetterException is raised exclusively by
+            // ValidateResourcesAcquiredEvent, before the processing loop — so ResourcesNormalized was
+            // provably never produced and nothing downstream can be holding {correlationId}. If a
+            // DeadLetterException is ever thrown after the produce, this must become
+            // AcquisitionKeysOnly.
             await _resourceCachePurger.PurgeAsync(
                 result.Message.Value,
                 $"{nameof(KafkaTopic.ResourcesAcquired)} dead-lettered: {ex.Message}",
+                ResourceCachePurgeScope.All,
                 consumeCancellationToken);
         }
         catch (TransientException ex)

DotNet/ServiceTests/UnitTests/Normalization/ResourceCachePurgerTests.cs

@@ -14,7 +14,7 @@ public class ResourceCachePurgerTests
     private const string CorrelationId = "3fa85f64-5717-4562-b3fc-2c963f66afa6";
 
     [Fact]
-    public async Task PurgeAsync_DeletesAcquisitionKeysAndCorrelationKey()
+    public async Task PurgeAsync_ScopeAll_DeletesAcquisitionKeysAndCorrelationKey()
     {
         var (purger, cache, implementation) = BuildPurger(ResourceCacheType.Redis);
 
@@ -24,7 +24,7 @@ public async Task PurgeAsync_DeletesAcquisitionKeysAndCorrelationKey()
             .Callback<List<string>, CancellationToken>((keys, _) => deletedKeys = keys)
             .Returns(Task.CompletedTask);
 
-        await purger.PurgeAsync(BuildValue(ResourceCacheType.Redis), "test");
+        await purger.PurgeAsync(BuildValue(ResourceCacheType.Redis), "test", ResourceCachePurgeScope.All);
 
         cache.Verify(item => item.GetImplementation(ResourceCacheType.Redis), Times.Once);
         Assert.NotNull(deletedKeys);
@@ -33,6 +33,30 @@ public async Task PurgeAsync_DeletesAcquisitionKeysAndCorrelationKey()
             deletedKeys);
     }
 
+    [Fact]
+    public async Task PurgeAsync_ScopeAcquisitionKeysOnly_LeavesTheCorrelationKey()
+    {
+        // The retry-exhausted path cannot prove that an earlier attempt did not already publish
+        // ResourcesNormalized (the produce precedes the acquisition-key delete on the success path),
+        // so Measure Eval may be holding {correlationId} for its SUPPLEMENTAL pass. That key must
+        // survive this purge; the cache expiration policy reclaims it if nothing needed it.
+        var (purger, _, implementation) = BuildPurger(ResourceCacheType.Redis);
+
+        List<string>? deletedKeys = null;
+        implementation
+            .Setup(item => item.DeleteAsync(It.IsAny<List<string>>(), It.IsAny<CancellationToken>()))
+            .Callback<List<string>, CancellationToken>((keys, _) => deletedKeys = keys)
+            .Returns(Task.CompletedTask);
+
+        await purger.PurgeAsync(BuildValue(ResourceCacheType.Redis), "test", ResourceCachePurgeScope.AcquisitionKeysOnly);
+
+        Assert.NotNull(deletedKeys);
+        Assert.Equal(
+            new List<string> { $"{CorrelationId}:Patient", $"{CorrelationId}:Encounter" },
+            deletedKeys);
+        Assert.DoesNotContain(CorrelationId, deletedKeys!);
+    }
+
     [Fact]
     public async Task PurgeAsync_UsesTheCacheTypeCarriedOnTheMessage()
     {
@@ -42,7 +66,7 @@ public async Task PurgeAsync_UsesTheCacheTypeCarriedOnTheMessage()
             .Setup(item => item.DeleteAsync(It.IsAny<List<string>>(), It.IsAny<CancellationToken>()))
             .Returns(Task.CompletedTask);
 
-        await purger.PurgeAsync(BuildValue(ResourceCacheType.ABS), "test");
+        await purger.PurgeAsync(BuildValue(ResourceCacheType.ABS), "test", ResourceCachePurgeScope.All);
 
         cache.Verify(item => item.GetImplementation(ResourceCacheType.ABS), Times.Once);
     }
@@ -61,7 +85,7 @@ public async Task PurgeAsync_DoesNotDeleteTwiceWhenTheCorrelationKeyIsAlreadyPre
         var value = BuildValue(ResourceCacheType.Redis);
         value.CacheKeys.Add(CorrelationId);
 
-        await purger.PurgeAsync(value, "test");
+        await purger.PurgeAsync(value, "test", ResourceCachePurgeScope.All);
 
         Assert.NotNull(deletedKeys);
         Assert.Equal(deletedKeys!.Count, deletedKeys.Distinct().Count());
@@ -80,7 +104,7 @@ private static async Task AssertNoDelete(List<string>? cacheKeys)
         var value = BuildValue(ResourceCacheType.Redis);
         value.CacheKeys = cacheKeys!;
 
-        await purger.PurgeAsync(value, "test");
+        await purger.PurgeAsync(value, "test", ResourceCachePurgeScope.All);
 
         cache.Verify(item => item.GetImplementation(It.IsAny<ResourceCacheType>()), Times.Never);
         implementation.Verify(
@@ -92,7 +116,7 @@ public async Task PurgeAsync_WithNullValue_DoesNotThrow()
     {
         var (purger, cache, _) = BuildPurger(ResourceCacheType.Redis);
 
-        await purger.PurgeAsync(null, "test");
+        await purger.PurgeAsync(null, "test", ResourceCachePurgeScope.All);
 
         cache.Verify(item => item.GetImplementation(It.IsAny<ResourceCacheType>()), Times.Never);
     }
@@ -107,7 +131,7 @@ public async Task PurgeAsync_WhenDeleteThrows_SwallowsTheException()
             .ThrowsAsync(new InvalidOperationException("cache unavailable"));
 
         // The caller is already handling a failed message; cleanup failure must not add another.
-        await purger.PurgeAsync(BuildValue(ResourceCacheType.Redis), "test");
+        await purger.PurgeAsync(BuildValue(ResourceCacheType.Redis), "test", ResourceCachePurgeScope.All);
     }
 
     private static (ResourceCachePurger, Mock<IResourceCache>, Mock<IResourceCache>) BuildPurger(ResourceCacheType cacheType)

DotNet/ServiceTests/UnitTests/Normalization/ResourcesAcquiredListenerCacheCleanupTests.cs

@@ -44,8 +44,10 @@ public async Task ConsumeMessageAsync_DeadLetterFailure_PurgesTheResourceCache()
 
         deadLetterHandler.Verify(
             item => item.HandleException(result, It.IsAny<DeadLetterException>(), FacilityId), Times.Once);
+        // The immediate dead-letter path provably fails before ResourcesNormalized is produced
+        // (DeadLetterException is raised only by validation), so the full purge is safe here.
         purger.Verify(
-            item => item.PurgeAsync(result.Message.Value, It.IsAny<string>(), It.IsAny<CancellationToken>()),
+            item => item.PurgeAsync(result.Message.Value, It.IsAny<string>(), ResourceCachePurgeScope.All, It.IsAny<CancellationToken>()),
             Times.Once);
     }
 

DotNet/ServiceTests/UnitTests/Normalization/ResourcesAcquiredRetryDeadLetterHandlerTests.cs

@@ -40,9 +40,10 @@ public void ProduceDeadLetter_PurgesTheCacheKeysCarriedOnTheRetriedMessage()
         };
 
         ResourcesAcquiredValue? purged = null;
+        ResourceCachePurgeScope? scope = null;
         purger
-            .Setup(item => item.PurgeAsync(It.IsAny<ResourcesAcquiredValue>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
-            .Callback<ResourcesAcquiredValue?, string, CancellationToken>((v, _, _) => purged = v)
+            .Setup(item => item.PurgeAsync(It.IsAny<ResourcesAcquiredValue>(), It.IsAny<string>(), It.IsAny<ResourceCachePurgeScope>(), It.IsAny<CancellationToken>()))
+            .Callback<ResourcesAcquiredValue?, string, ResourceCachePurgeScope, CancellationToken>((v, _, s, _) => { purged = v; scope = s; })
             .Returns(Task.CompletedTask);
 
         handler.ProduceDeadLetter(BuildRetryConsumeResult(JsonSerializer.Serialize(value)), "Retry count exceeded");
@@ -51,6 +52,10 @@ public void ProduceDeadLetter_PurgesTheCacheKeysCarriedOnTheRetriedMessage()
         Assert.Equal(value.CacheKeys, purged!.CacheKeys);
         Assert.Equal(ResourceCacheType.ABS, purged.CacheType);
 
+        // Retry exhaustion cannot prove an earlier attempt did not already publish
+        // ResourcesNormalized, so it must never remove {correlationId}.
+        Assert.Equal(ResourceCachePurgeScope.AcquisitionKeysOnly, scope);
+
         // The dead letter is still produced: the durable record of the failure comes first.
         producer.Verify(
             item => item.Produce(