Skip to content

Commit 55fc5b0

Browse files
arianamihailescuarianamihailescu
committed
Merge branch 'user/steven.williams/LEGLINK-958' into LEGLINK-186-KafkaRetryImplementation
2 parents d2d81c2 + 7f6fcf8 commit 55fc5b0

11 files changed

Lines changed: 830 additions & 40 deletions

File tree

‎.github/workflows/tests.yaml‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -297,12 +297,23 @@ jobs:
297297
${{ runner.os }}-nuget-
298298
299299
# Pulled up-front so the integration fixtures (Testcontainers) don't
300-
# spend their per-test pull budget on the cold pull.
300+
# spend their per-test pull budget on the cold pull. Retry up to 3 times
301+
# to handle transient MCR network timeouts.
301302
- name: Pull Azurite Docker image
302-
run: docker pull mcr.microsoft.com/azure-storage/azurite:latest
303+
run: |
304+
for i in 1 2 3; do
305+
docker pull mcr.microsoft.com/azure-storage/azurite:latest && break
306+
echo "Pull attempt $i failed, retrying in 15s..."
307+
sleep 15
308+
done
303309
304310
- name: Pull SQL Server Docker image
305-
run: docker pull mcr.microsoft.com/mssql/server:2022-CU13-ubuntu-22.04
311+
run: |
312+
for i in 1 2 3; do
313+
docker pull mcr.microsoft.com/mssql/server:2022-CU13-ubuntu-22.04 && break
314+
echo "Pull attempt $i failed, retrying in 15s..."
315+
sleep 15
316+
done
306317
307318
- name: Restore dependencies
308319
run: dotnet restore DotNet/ServiceTests/ServiceTests.csproj

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

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,90 @@
1+
using Confluent.Kafka;
2+
using LantanaGroup.Link.Normalization.Application.Models.Messages;
3+
using LantanaGroup.Link.Normalization.Application.Services;
4+
using LantanaGroup.Link.Shared.Application.Error.Handlers;
5+
using LantanaGroup.Link.Shared.Application.Error.Interfaces;
6+
using LantanaGroup.Link.Shared.Application.Interfaces;
7+
using LantanaGroup.Link.Shared.Application.Listeners;
8+
using LantanaGroup.Link.Shared.Application.Models;
9+
using LantanaGroup.Link.Shared.Application.Services.Security;
10+
using System.Text.Json;
11+
12+
namespace LantanaGroup.Link.Normalization.Application.Error;
13+
14+
/// <summary>
15+
/// Dead letter handler for <see cref="RetryListener"/> that also releases the resource cache.
16+
/// </summary>
17+
/// <remarks>
18+
/// A <c>ResourcesAcquired</c> message that fails transiently is republished to
19+
/// <c>ResourcesAcquired-Retry</c> and redelivered on a schedule. Once the retry count is exhausted,
20+
/// <see cref="RetryListener"/> — shared, service-agnostic code with no knowledge of the resource
21+
/// cache — dead-letters it to <c>ResourcesAcquired-Error</c>. That is the second and final terminal
22+
/// path for the message (the first being a <c>DeadLetterException</c> raised directly in
23+
/// <c>ResourcesAcquiredListener</c>), so it is where the cached resources have to be released.
24+
/// <para>
25+
/// Normalization registers <see cref="RetryListener"/> for <c>ResourcesAcquired-Retry</c> only, so
26+
/// every message reaching this handler is a <see cref="ResourcesAcquiredValue"/>.
27+
/// </para>
28+
/// </remarks>
29+
public class ResourcesAcquiredRetryDeadLetterHandler : DeadLetterExceptionHandler<RetryListener, string, string>
30+
{
31+
private static readonly JsonSerializerOptions DeserializerOptions = new()
32+
{
33+
// Mirrors JsonWithFhirMessageDeserializer, which is how the value was read off the topic.
34+
PropertyNameCaseInsensitive = true,
35+
AllowTrailingCommas = true
36+
};
37+
38+
private readonly IResourceCachePurger _resourceCachePurger;
39+
private readonly ILogger<ResourcesAcquiredRetryDeadLetterHandler> _logger;
40+
41+
public ResourcesAcquiredRetryDeadLetterHandler(
42+
IKafkaProducerFactory<string, string> producerFactory,
43+
IKafkaProducerFactory<string, string> nullConsumeResultProducerFactory,
44+
ServiceInformation serviceInformation,
45+
IExceptionLogger<RetryListener> exceptionHandler,
46+
IResourceCachePurger resourceCachePurger,
47+
ILogger<ResourcesAcquiredRetryDeadLetterHandler> logger)
48+
: base(producerFactory, nullConsumeResultProducerFactory, serviceInformation, exceptionHandler)
49+
{
50+
_resourceCachePurger = resourceCachePurger ?? throw new ArgumentNullException(nameof(resourceCachePurger));
51+
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
52+
}
53+
54+
/// <remarks>
55+
/// Both <c>HandleException</c> overloads funnel through here. The dead letter is produced first so
56+
/// that the durable record of the failure is never at the mercy of the cache purge.
57+
/// </remarks>
58+
public override void ProduceDeadLetter(ConsumeResult<string, string> consumeResult, string exceptionMessage)
59+
{
60+
base.ProduceDeadLetter(consumeResult, exceptionMessage);
61+
62+
PurgeResourceCache(consumeResult, exceptionMessage);
63+
}
64+
65+
private void PurgeResourceCache(ConsumeResult<string, string> consumeResult, string exceptionMessage)
66+
{
67+
ResourcesAcquiredValue? value;
68+
69+
try
70+
{
71+
value = JsonSerializer.Deserialize<ResourcesAcquiredValue>(
72+
consumeResult.Message.Value, DeserializerOptions);
73+
}
74+
catch (Exception ex)
75+
{
76+
_logger.LogError(ex,
77+
"Could not deserialize a retry-exhausted message from {Topic} to determine its resource cache keys. " +
78+
"Any cached resources for it will be released by the cache expiration policy instead.",
79+
consumeResult.Topic.SanitizeForLog());
80+
return;
81+
}
82+
83+
// RetryListener runs its consume callback on a background thread with no synchronization
84+
// context, so blocking here cannot deadlock. PurgeAsync does not throw.
85+
_resourceCachePurger
86+
.PurgeAsync(value, $"retry count exhausted: {exceptionMessage}")
87+
.GetAwaiter()
88+
.GetResult();
89+
}
90+
}

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

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
using LantanaGroup.Link.Normalization.Application.Models.Messages;
2+
using LantanaGroup.Link.Shared.Application.Interfaces;
3+
using LantanaGroup.Link.Shared.Application.Services.Security;
4+
5+
namespace LantanaGroup.Link.Normalization.Application.Services;
6+
7+
/// <summary>
8+
/// Releases the resource cache entries belonging to a <c>ResourcesAcquired</c> message after a
9+
/// terminal (non-retryable) normalization failure.
10+
/// </summary>
11+
/// <remarks>
12+
/// Only call this on terminal failures. A message bound for <c>ResourcesAcquired-Retry</c> still
13+
/// needs its cached resources when it is redelivered, so purging on a transient failure would
14+
/// guarantee the retry fails too.
15+
/// <para>
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.
23+
/// </para>
24+
/// </remarks>
25+
public interface IResourceCachePurger
26+
{
27+
Task PurgeAsync(ResourcesAcquiredValue? value, string reason, CancellationToken cancellationToken = default);
28+
}
29+
30+
/// <inheritdoc cref="IResourceCachePurger"/>
31+
public class ResourceCachePurger : IResourceCachePurger
32+
{
33+
private readonly IResourceCache _resourceCache;
34+
private readonly ILogger<ResourceCachePurger> _logger;
35+
36+
public ResourceCachePurger(IResourceCache resourceCache, ILogger<ResourceCachePurger> logger)
37+
{
38+
_resourceCache = resourceCache ?? throw new ArgumentNullException(nameof(resourceCache));
39+
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
40+
}
41+
42+
public async Task PurgeAsync(ResourcesAcquiredValue? value, string reason, CancellationToken cancellationToken = default)
43+
{
44+
var cacheKeys = value?.CacheKeys?.Where(k => !string.IsNullOrWhiteSpace(k)).ToList();
45+
46+
if (cacheKeys == null || cacheKeys.Count == 0)
47+
{
48+
_logger.LogWarning(
49+
"Cannot purge resource cache after terminal failure ({Reason}): the message carries no cache keys. " +
50+
"Any cached resources for it will be released by the cache expiration policy instead.",
51+
reason.SanitizeForLog());
52+
return;
53+
}
54+
55+
try
56+
{
57+
await _resourceCache
58+
.GetImplementation(value!.CacheType)
59+
.DeleteAsync(cacheKeys, cancellationToken);
60+
61+
_logger.LogInformation(
62+
"Purged {KeyCount} resource cache entries after terminal failure ({Reason}). CacheType: {CacheType}, Keys: [{Keys}]",
63+
cacheKeys.Count,
64+
reason.SanitizeForLog(),
65+
value.CacheType,
66+
string.Join(", ", cacheKeys).SanitizeForLog());
67+
}
68+
catch (Exception ex)
69+
{
70+
// Never let cleanup failure escape: the caller is already handling a failed message, and
71+
// the cache expiration policy is the backstop for whatever we could not delete here.
72+
_logger.LogError(ex,
73+
"Failed to purge resource cache after terminal failure ({Reason}). CacheType: {CacheType}, Keys: [{Keys}]",
74+
reason.SanitizeForLog(),
75+
value!.CacheType,
76+
string.Join(", ", cacheKeys).SanitizeForLog());
77+
}
78+
}
79+
}

‎DotNet/Normalization/Listeners/ResourcesAcquiredListener.cs‎

Lines changed: 47 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@ public class ResourcesAcquiredListener : BackgroundService
4444
private readonly CopyLocationAliasToTypeIterativelyOperationService _copyLocationAliasToTypeIterativelyOperationService;
4545
private readonly RemoveExtensionsOperationService _removeExtensionsOperationService;
4646
private readonly IResourceCache _resourceCache;
47+
private readonly IResourceCachePurger _resourceCachePurger;
4748

4849
public ResourcesAcquiredListener(
4950
ILogger<ResourcesAcquiredListener> logger,
@@ -61,7 +62,8 @@ public ResourcesAcquiredListener(
6162
CopyLocationOperationService copyLocationOperationService,
6263
CopyLocationAliasToTypeIterativelyOperationService copyLocationAliasToTypeIterativelyOperationService,
6364
RemoveExtensionsOperationService removeExtensionsOperationService,
64-
IResourceCache resourceCache)
65+
IResourceCache resourceCache,
66+
IResourceCachePurger resourceCachePurger)
6567
{
6668
this._logger = logger ?? throw new ArgumentNullException(nameof(logger));
6769
_consumerFactory = consumerFactory ?? throw new ArgumentNullException(nameof(consumerFactory));
@@ -88,6 +90,7 @@ public ResourcesAcquiredListener(
8890
_copyLocationAliasToTypeIterativelyOperationService = copyLocationAliasToTypeIterativelyOperationService ?? throw new ArgumentNullException(nameof(copyLocationAliasToTypeIterativelyOperationService));
8991
_removeExtensionsOperationService = removeExtensionsOperationService ?? throw new ArgumentNullException(nameof(removeExtensionsOperationService));
9092
_resourceCache = resourceCache ?? throw new ArgumentNullException(nameof(resourceCache));
93+
_resourceCachePurger = resourceCachePurger ?? throw new ArgumentNullException(nameof(resourceCachePurger));
9194
}
9295

9396
protected override async Task ExecuteAsync(CancellationToken cancellationToken)
@@ -113,25 +116,7 @@ await kafkaConsumer.ConsumeWithInstrumentation(async (result, consumeCancellatio
113116
{
114117
try
115118
{
116-
await ProcessMessageAsync(result, consumeCancellationToken);
117-
}
118-
catch (DeadLetterException ex)
119-
{
120-
_deadLetterExceptionHandler.HandleException(result, ex, result.Message.Key?.FacilityId ?? string.Empty);
121-
}
122-
catch (TransientException ex)
123-
{
124-
_transientExceptionHandler.HandleException(result, ex, result.Message.Key?.FacilityId ?? string.Empty);
125-
}
126-
catch (OperationCanceledException) when (consumeCancellationToken.IsCancellationRequested)
127-
{
128-
throw;
129-
}
130-
catch (Exception ex)
131-
{
132-
_logger.LogError(ex, "Failed to process ResourceAcquired event for facility {FacilityId}.", result?.Message.Key?.FacilityId?.SanitizeForLog());
133-
134-
_transientExceptionHandler.HandleException(result, new TransientException("Normalization Exception thrown: " + ex.Message, ex), result.Message.Key?.FacilityId ?? string.Empty);
119+
await ConsumeMessageAsync(result, consumeCancellationToken);
135120
}
136121
finally
137122
{
@@ -182,6 +167,48 @@ await kafkaConsumer.ConsumeWithInstrumentation(async (result, consumeCancellatio
182167
}
183168
}
184169

170+
/// <summary>
171+
/// Processes a single consumed message and routes any failure to the dead letter or retry topic.
172+
/// </summary>
173+
/// <remarks>
174+
/// Separate from the consume loop (which owns only the offset commit) so that the failure routing —
175+
/// in particular which failures release the resource cache — is directly testable.
176+
/// </remarks>
177+
public async Task ConsumeMessageAsync(ConsumeResult<ResourceKey, ResourcesAcquiredValue> result, CancellationToken consumeCancellationToken)
178+
{
179+
try
180+
{
181+
await ProcessMessageAsync(result, consumeCancellationToken);
182+
}
183+
catch (DeadLetterException ex)
184+
{
185+
_deadLetterExceptionHandler.HandleException(result, ex, result.Message.Key?.FacilityId ?? string.Empty);
186+
187+
// Terminal failure: the message is on ResourcesAcquired-Error and will never be normalized,
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.
191+
await _resourceCachePurger.PurgeAsync(
192+
result.Message.Value,
193+
$"{nameof(KafkaTopic.ResourcesAcquired)} dead-lettered: {ex.Message}",
194+
consumeCancellationToken);
195+
}
196+
catch (TransientException ex)
197+
{
198+
_transientExceptionHandler.HandleException(result, ex, result.Message.Key?.FacilityId ?? string.Empty);
199+
}
200+
catch (OperationCanceledException) when (consumeCancellationToken.IsCancellationRequested)
201+
{
202+
throw;
203+
}
204+
catch (Exception ex)
205+
{
206+
_logger.LogError(ex, "Failed to process ResourceAcquired event for facility {FacilityId}.", result?.Message.Key?.FacilityId?.SanitizeForLog());
207+
208+
_transientExceptionHandler.HandleException(result, new TransientException("Normalization Exception thrown: " + ex.Message, ex), result.Message.Key?.FacilityId ?? string.Empty);
209+
}
210+
}
211+
185212
public async Task ProcessMessageAsync(ConsumeResult<ResourceKey, ResourcesAcquiredValue> result, CancellationToken cancellationToken)
186213
{
187214
ValidateResourcesAcquiredEvent(result, out string correlationId);

‎DotNet/Normalization/Program.cs‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
using HealthChecks.UI.Client;
44
using Hl7.Fhir.Model.CdsHooks;
55
using LantanaGroup.Link.Shared.Application.Models.Configs;
6+
using LantanaGroup.Link.Normalization.Application.Error;
67
using LantanaGroup.Link.Normalization.Application.Models.Messages;
78
using LantanaGroup.Link.Normalization.Application.Services;
89
using LantanaGroup.Link.Normalization.Application.Services.Operations;
@@ -97,6 +98,14 @@ static void RegisterServices(WebApplicationBuilder builder)
9798
builder.Services.AddSingleton(typeof(ITransientExceptionHandler<,,>), typeof(TransientExceptionHandler<,,>));
9899
builder.Services.AddSingleton(typeof(IDeadLetterExceptionHandler<,,>), typeof(DeadLetterExceptionHandler<,,>));
99100

101+
builder.Services.AddSingleton<IResourceCachePurger, ResourceCachePurger>();
102+
103+
// A closed-type registration takes precedence over the open generic above regardless of the order
104+
// the two appear in, so this wins for IDeadLetterExceptionHandler<RetryListener, string, string>:
105+
// when RetryListener exhausts the retry count for a ResourcesAcquired message, the dead letter also
106+
// releases the resource cache. See ResourcesAcquiredRetryDeadLetterHandler.
107+
builder.Services.AddSingleton<IDeadLetterExceptionHandler<RetryListener, string, string>, ResourcesAcquiredRetryDeadLetterHandler>();
108+
100109
builder.Services.AddTransient<ITenantApiService, TenantApiService>();
101110

102111
builder.Services.AddControllers()

‎DotNet/ServiceTests/IntegrationTests/Normalization/NormalizationIntegrationTestFixture.cs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -159,6 +159,7 @@ public async Task InitializeAsync()
159159
builder.Services.AddKeyedSingleton<IResourceCache, RedisResourceCache>(ResourceCacheType.Redis);
160160
builder.Services.AddKeyedSingleton<IResourceCache, ABSResourceCache>(ResourceCacheType.ABS);
161161
builder.Services.AddSingleton<IResourceCache, HybridResourceCache>();
162+
builder.Services.AddSingleton<IResourceCachePurger, ResourceCachePurger>();
162163

163164
builder.Services.AddMemoryCache();
164165

0 commit comments

Comments
 (0)