Skip to content

Commit 7f6fcf8

Browse files
arianamihailescuarianamihailescu
committed
LEGLINK-958: Normalization never deletes the correlation key
Replace the ResourceCachePurgeScope enum with a single ownership rule: Normalization deletes only its own input, the {correlationId}:{ResourceType} acquisition keys. The {correlationId} key belongs to its reader - Measure Eval deletes it after evaluation (keepCacheForSupplemental aside), and the cache expiration policy reclaims keys whose message never got published. Scope.All on the immediate dead-letter path was unsafe for SUPPLEMENTAL messages: DataAcquisition reuses the correlationId across phases, so after a reportable INITIAL pass the key holds kept, unrebuildable INITIAL data. A malformed SUPPLEMENTAL message would have destroyed it and turned the eventual evaluation into a silent false not-reportable. Its safety also rested on a comment-enforced invariant (DeadLetterException is only thrown before the produce) that no compiler or test could defend. Removing the option instead of gating it deletes the enum, the scope parameter, the correlation-id derivation, and the invariant comments. The retry-exhausted path behaves exactly as before; the dead-letter path now leaves {correlationId} to its owner. Claude-Session: https://claude.ai/code/session_01KN565tFAuJUAEkK2DdRF1e
1 parent ad9a7b9 commit 7f6fcf8

6 files changed

Lines changed: 29 additions & 129 deletions

File tree

‎DotNet/Normalization/Application/Error/ResourcesAcquiredRetryDeadLetterHandler.cs‎

Lines changed: 1 addition & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -80,16 +80,10 @@ private void PurgeResourceCache(ConsumeResult<string, string> consumeResult, str
8080
return;
8181
}
8282

83-
// Retry exhaustion cannot rule out that an earlier attempt already produced
84-
// ResourcesNormalized and then failed on the trailing cache delete — the produce precedes
85-
// the delete on the success path — in which case Measure Eval is holding {correlationId}
86-
// for its SUPPLEMENTAL pass. Deleting it would make that pass read an empty cache and emit
87-
// a silent false not-reportable report, so only the acquisition keys are released here.
88-
//
8983
// RetryListener runs its consume callback on a background thread with no synchronization
9084
// context, so blocking here cannot deadlock. PurgeAsync does not throw.
9185
_resourceCachePurger
92-
.PurgeAsync(value, $"retry count exhausted: {exceptionMessage}", ResourceCachePurgeScope.AcquisitionKeysOnly)
86+
.PurgeAsync(value, $"retry count exhausted: {exceptionMessage}")
9387
.GetAwaiter()
9488
.GetResult();
9589
}

‎DotNet/Normalization/Application/Services/ResourceCachePurger.cs‎

Lines changed: 13 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -4,30 +4,6 @@
44

55
namespace LantanaGroup.Link.Normalization.Application.Services;
66

7-
/// <summary>
8-
/// Which cache keys a terminal-failure purge is allowed to remove. The scope encodes what the
9-
/// caller can prove about whether <c>ResourcesNormalized</c> was already published for the message.
10-
/// </summary>
11-
public enum ResourceCachePurgeScope
12-
{
13-
/// <summary>
14-
/// Remove the per-resource-type acquisition keys and the <c>{correlationId}</c> key. Only valid
15-
/// where the failure provably occurred before <c>ResourcesNormalized</c> was produced (the
16-
/// immediate dead-letter path: validation raises before the processing loop), so nothing
17-
/// downstream can be holding the correlation key.
18-
/// </summary>
19-
All,
20-
21-
/// <summary>
22-
/// Remove only the per-resource-type acquisition keys. For paths that cannot rule out a prior
23-
/// publish — retry exhaustion follows attempts that may have produced <c>ResourcesNormalized</c>
24-
/// and then failed on the trailing cache delete — where Measure Eval may already be holding
25-
/// <c>{correlationId}</c> for its SUPPLEMENTAL pass. The correlation key is left to the cache
26-
/// expiration policy.
27-
/// </summary>
28-
AcquisitionKeysOnly
29-
}
30-
317
/// <summary>
328
/// Releases the resource cache entries belonging to a <c>ResourcesAcquired</c> message after a
339
/// terminal (non-retryable) normalization failure.
@@ -37,15 +13,18 @@ public enum ResourceCachePurgeScope
3713
/// needs its cached resources when it is redelivered, so purging on a transient failure would
3814
/// guarantee the retry fails too.
3915
/// <para>
40-
/// The success path deletes only the per-resource-type acquisition keys and leaves
41-
/// <c>{correlationId}</c> in place for Measure Eval to read. Whether a terminal failure may also
42-
/// remove the correlation key depends on what the caller can prove — see
43-
/// <see cref="ResourceCachePurgeScope"/>.
16+
/// Only the per-resource-type acquisition keys (<c>{correlationId}:{ResourceType}</c>) are ever
17+
/// deleted — they are Normalization's input, consumed by this message alone. The
18+
/// <c>{correlationId}</c> key is Normalization's output and belongs to its reader: Measure Eval
19+
/// deletes it after evaluation, and the cache expiration policy reclaims it if no reader ever
20+
/// comes (e.g. the failure occurred before <c>ResourcesNormalized</c> was published). Normalization
21+
/// never deletes it, on any path — it cannot know whether an earlier attempt already published, or
22+
/// whether the key still holds kept INITIAL-phase data a SUPPLEMENTAL evaluation needs.
4423
/// </para>
4524
/// </remarks>
4625
public interface IResourceCachePurger
4726
{
48-
Task PurgeAsync(ResourcesAcquiredValue? value, string reason, ResourceCachePurgeScope scope, CancellationToken cancellationToken = default);
27+
Task PurgeAsync(ResourcesAcquiredValue? value, string reason, CancellationToken cancellationToken = default);
4928
}
5029

5130
/// <inheritdoc cref="IResourceCachePurger"/>
@@ -60,7 +39,7 @@ public ResourceCachePurger(IResourceCache resourceCache, ILogger<ResourceCachePu
6039
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
6140
}
6241

63-
public async Task PurgeAsync(ResourcesAcquiredValue? value, string reason, ResourceCachePurgeScope scope, CancellationToken cancellationToken = default)
42+
public async Task PurgeAsync(ResourcesAcquiredValue? value, string reason, CancellationToken cancellationToken = default)
6443
{
6544
var cacheKeys = value?.CacheKeys?.Where(k => !string.IsNullOrWhiteSpace(k)).ToList();
6645

@@ -73,31 +52,18 @@ public async Task PurgeAsync(ResourcesAcquiredValue? value, string reason, Resou
7352
return;
7453
}
7554

76-
// Acquisition keys are "{correlationId}:{ResourceType}". The correlation key itself is only
77-
// removed when the caller can prove ResourcesNormalized was never published for this message
78-
// (scope All); otherwise Measure Eval may still be reading it, and it is left to expire.
79-
var keysToDelete = new List<string>(cacheKeys);
80-
if (scope == ResourceCachePurgeScope.All)
81-
{
82-
keysToDelete.AddRange(cacheKeys
83-
.Select(ExtractCorrelationId)
84-
.Where(correlationId => !string.IsNullOrWhiteSpace(correlationId))
85-
.Distinct()
86-
.Where(correlationId => !keysToDelete.Contains(correlationId)));
87-
}
88-
8955
try
9056
{
9157
await _resourceCache
9258
.GetImplementation(value!.CacheType)
93-
.DeleteAsync(keysToDelete, cancellationToken);
59+
.DeleteAsync(cacheKeys, cancellationToken);
9460

9561
_logger.LogInformation(
9662
"Purged {KeyCount} resource cache entries after terminal failure ({Reason}). CacheType: {CacheType}, Keys: [{Keys}]",
97-
keysToDelete.Count,
63+
cacheKeys.Count,
9864
reason.SanitizeForLog(),
9965
value.CacheType,
100-
string.Join(", ", keysToDelete).SanitizeForLog());
66+
string.Join(", ", cacheKeys).SanitizeForLog());
10167
}
10268
catch (Exception ex)
10369
{
@@ -107,14 +73,7 @@ await _resourceCache
10773
"Failed to purge resource cache after terminal failure ({Reason}). CacheType: {CacheType}, Keys: [{Keys}]",
10874
reason.SanitizeForLog(),
10975
value!.CacheType,
110-
string.Join(", ", keysToDelete).SanitizeForLog());
76+
string.Join(", ", cacheKeys).SanitizeForLog());
11177
}
11278
}
113-
114-
/// <remarks>Mirrors <c>HybridResourceCache.ExtractCorrelationId</c>.</remarks>
115-
private static string ExtractCorrelationId(string cacheKey)
116-
{
117-
var idx = cacheKey.IndexOf(':');
118-
return idx > 0 ? cacheKey[..idx] : cacheKey;
119-
}
12079
}

‎DotNet/Normalization/Listeners/ResourcesAcquiredListener.cs‎

Lines changed: 3 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -185,18 +185,12 @@ public async Task ConsumeMessageAsync(ConsumeResult<ResourceKey, ResourcesAcquir
185185
_deadLetterExceptionHandler.HandleException(result, ex, result.Message.Key?.FacilityId ?? string.Empty);
186186

187187
// Terminal failure: the message is on ResourcesAcquired-Error and will never be normalized,
188-
// so release its cached resources. The retry paths below must NOT do this — a redelivered
189-
// message still needs its cache.
190-
//
191-
// Scope.All is safe here ONLY because DeadLetterException is raised exclusively by
192-
// ValidateResourcesAcquiredEvent, before the processing loop — so ResourcesNormalized was
193-
// provably never produced and nothing downstream can be holding {correlationId}. If a
194-
// DeadLetterException is ever thrown after the produce, this must become
195-
// AcquisitionKeysOnly.
188+
// so release its acquisition keys. The retry paths below must NOT do this — a redelivered
189+
// message still needs its cache. The {correlationId} key is never deleted here: Measure
190+
// Eval owns its deletion, and the cache expiration policy reclaims it if unread.
196191
await _resourceCachePurger.PurgeAsync(
197192
result.Message.Value,
198193
$"{nameof(KafkaTopic.ResourcesAcquired)} dead-lettered: {ex.Message}",
199-
ResourceCachePurgeScope.All,
200194
consumeCancellationToken);
201195
}
202196
catch (TransientException ex)

‎DotNet/ServiceTests/UnitTests/Normalization/ResourceCachePurgerTests.cs‎

Lines changed: 9 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -14,8 +14,11 @@ public class ResourceCachePurgerTests
1414
private const string CorrelationId = "3fa85f64-5717-4562-b3fc-2c963f66afa6";
1515

1616
[Fact]
17-
public async Task PurgeAsync_ScopeAll_DeletesAcquisitionKeysAndCorrelationKey()
17+
public async Task PurgeAsync_DeletesOnlyTheAcquisitionKeys_NeverTheCorrelationKey()
1818
{
19+
// Normalization only ever deletes its own input, the {correlationId}:{ResourceType}
20+
// acquisition keys. The {correlationId} key belongs to its reader: Measure Eval deletes it
21+
// after evaluation, and the cache expiration policy reclaims it if no reader ever comes.
1922
var (purger, cache, implementation) = BuildPurger(ResourceCacheType.Redis);
2023

2124
List<string>? deletedKeys = null;
@@ -24,32 +27,9 @@ public async Task PurgeAsync_ScopeAll_DeletesAcquisitionKeysAndCorrelationKey()
2427
.Callback<List<string>, CancellationToken>((keys, _) => deletedKeys = keys)
2528
.Returns(Task.CompletedTask);
2629

27-
await purger.PurgeAsync(BuildValue(ResourceCacheType.Redis), "test", ResourceCachePurgeScope.All);
30+
await purger.PurgeAsync(BuildValue(ResourceCacheType.Redis), "test");
2831

2932
cache.Verify(item => item.GetImplementation(ResourceCacheType.Redis), Times.Once);
30-
Assert.NotNull(deletedKeys);
31-
Assert.Equal(
32-
new List<string> { $"{CorrelationId}:Patient", $"{CorrelationId}:Encounter", CorrelationId },
33-
deletedKeys);
34-
}
35-
36-
[Fact]
37-
public async Task PurgeAsync_ScopeAcquisitionKeysOnly_LeavesTheCorrelationKey()
38-
{
39-
// The retry-exhausted path cannot prove that an earlier attempt did not already publish
40-
// ResourcesNormalized (the produce precedes the acquisition-key delete on the success path),
41-
// so Measure Eval may be holding {correlationId} for its SUPPLEMENTAL pass. That key must
42-
// survive this purge; the cache expiration policy reclaims it if nothing needed it.
43-
var (purger, _, implementation) = BuildPurger(ResourceCacheType.Redis);
44-
45-
List<string>? deletedKeys = null;
46-
implementation
47-
.Setup(item => item.DeleteAsync(It.IsAny<List<string>>(), It.IsAny<CancellationToken>()))
48-
.Callback<List<string>, CancellationToken>((keys, _) => deletedKeys = keys)
49-
.Returns(Task.CompletedTask);
50-
51-
await purger.PurgeAsync(BuildValue(ResourceCacheType.Redis), "test", ResourceCachePurgeScope.AcquisitionKeysOnly);
52-
5333
Assert.NotNull(deletedKeys);
5434
Assert.Equal(
5535
new List<string> { $"{CorrelationId}:Patient", $"{CorrelationId}:Encounter" },
@@ -66,31 +46,11 @@ public async Task PurgeAsync_UsesTheCacheTypeCarriedOnTheMessage()
6646
.Setup(item => item.DeleteAsync(It.IsAny<List<string>>(), It.IsAny<CancellationToken>()))
6747
.Returns(Task.CompletedTask);
6848

69-
await purger.PurgeAsync(BuildValue(ResourceCacheType.ABS), "test", ResourceCachePurgeScope.All);
49+
await purger.PurgeAsync(BuildValue(ResourceCacheType.ABS), "test");
7050

7151
cache.Verify(item => item.GetImplementation(ResourceCacheType.ABS), Times.Once);
7252
}
7353

74-
[Fact]
75-
public async Task PurgeAsync_DoesNotDeleteTwiceWhenTheCorrelationKeyIsAlreadyPresent()
76-
{
77-
var (purger, _, implementation) = BuildPurger(ResourceCacheType.Redis);
78-
79-
List<string>? deletedKeys = null;
80-
implementation
81-
.Setup(item => item.DeleteAsync(It.IsAny<List<string>>(), It.IsAny<CancellationToken>()))
82-
.Callback<List<string>, CancellationToken>((keys, _) => deletedKeys = keys)
83-
.Returns(Task.CompletedTask);
84-
85-
var value = BuildValue(ResourceCacheType.Redis);
86-
value.CacheKeys.Add(CorrelationId);
87-
88-
await purger.PurgeAsync(value, "test", ResourceCachePurgeScope.All);
89-
90-
Assert.NotNull(deletedKeys);
91-
Assert.Equal(deletedKeys!.Count, deletedKeys.Distinct().Count());
92-
}
93-
9454
[Fact]
9555
public async Task PurgeAsync_WithNullCacheKeys_DoesNotDelete() => await AssertNoDelete(null);
9656

@@ -104,7 +64,7 @@ private static async Task AssertNoDelete(List<string>? cacheKeys)
10464
var value = BuildValue(ResourceCacheType.Redis);
10565
value.CacheKeys = cacheKeys!;
10666

107-
await purger.PurgeAsync(value, "test", ResourceCachePurgeScope.All);
67+
await purger.PurgeAsync(value, "test");
10868

10969
cache.Verify(item => item.GetImplementation(It.IsAny<ResourceCacheType>()), Times.Never);
11070
implementation.Verify(
@@ -116,7 +76,7 @@ public async Task PurgeAsync_WithNullValue_DoesNotThrow()
11676
{
11777
var (purger, cache, _) = BuildPurger(ResourceCacheType.Redis);
11878

119-
await purger.PurgeAsync(null, "test", ResourceCachePurgeScope.All);
79+
await purger.PurgeAsync(null, "test");
12080

12181
cache.Verify(item => item.GetImplementation(It.IsAny<ResourceCacheType>()), Times.Never);
12282
}
@@ -131,7 +91,7 @@ public async Task PurgeAsync_WhenDeleteThrows_SwallowsTheException()
13191
.ThrowsAsync(new InvalidOperationException("cache unavailable"));
13292

13393
// The caller is already handling a failed message; cleanup failure must not add another.
134-
await purger.PurgeAsync(BuildValue(ResourceCacheType.Redis), "test", ResourceCachePurgeScope.All);
94+
await purger.PurgeAsync(BuildValue(ResourceCacheType.Redis), "test");
13595
}
13696

13797
private static (ResourceCachePurger, Mock<IResourceCache>, Mock<IResourceCache>) BuildPurger(ResourceCacheType cacheType)

‎DotNet/ServiceTests/UnitTests/Normalization/ResourcesAcquiredListenerCacheCleanupTests.cs‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -44,10 +44,8 @@ public async Task ConsumeMessageAsync_DeadLetterFailure_PurgesTheResourceCache()
4444

4545
deadLetterHandler.Verify(
4646
item => item.HandleException(result, It.IsAny<DeadLetterException>(), FacilityId), Times.Once);
47-
// The immediate dead-letter path provably fails before ResourcesNormalized is produced
48-
// (DeadLetterException is raised only by validation), so the full purge is safe here.
4947
purger.Verify(
50-
item => item.PurgeAsync(result.Message.Value, It.IsAny<string>(), ResourceCachePurgeScope.All, It.IsAny<CancellationToken>()),
48+
item => item.PurgeAsync(result.Message.Value, It.IsAny<string>(), It.IsAny<CancellationToken>()),
5149
Times.Once);
5250
}
5351

‎DotNet/ServiceTests/UnitTests/Normalization/ResourcesAcquiredRetryDeadLetterHandlerTests.cs‎

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -40,10 +40,9 @@ public void ProduceDeadLetter_PurgesTheCacheKeysCarriedOnTheRetriedMessage()
4040
};
4141

4242
ResourcesAcquiredValue? purged = null;
43-
ResourceCachePurgeScope? scope = null;
4443
purger
45-
.Setup(item => item.PurgeAsync(It.IsAny<ResourcesAcquiredValue>(), It.IsAny<string>(), It.IsAny<ResourceCachePurgeScope>(), It.IsAny<CancellationToken>()))
46-
.Callback<ResourcesAcquiredValue?, string, ResourceCachePurgeScope, CancellationToken>((v, _, s, _) => { purged = v; scope = s; })
44+
.Setup(item => item.PurgeAsync(It.IsAny<ResourcesAcquiredValue>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
45+
.Callback<ResourcesAcquiredValue?, string, CancellationToken>((v, _, _) => purged = v)
4746
.Returns(Task.CompletedTask);
4847

4948
handler.ProduceDeadLetter(BuildRetryConsumeResult(JsonSerializer.Serialize(value)), "Retry count exceeded");
@@ -52,10 +51,6 @@ public void ProduceDeadLetter_PurgesTheCacheKeysCarriedOnTheRetriedMessage()
5251
Assert.Equal(value.CacheKeys, purged!.CacheKeys);
5352
Assert.Equal(ResourceCacheType.ABS, purged.CacheType);
5453

55-
// Retry exhaustion cannot prove an earlier attempt did not already publish
56-
// ResourcesNormalized, so it must never remove {correlationId}.
57-
Assert.Equal(ResourceCachePurgeScope.AcquisitionKeysOnly, scope);
58-
5954
// The dead letter is still produced: the durable record of the failure comes first.
6055
producer.Verify(
6156
item => item.Produce(

0 commit comments

Comments
 (0)