fix(notifications): deterministic lowest-Id SMTP config selection (+ multi-row warning); atomic OAuth2 token cache entry

Claude-Session: https://claude.ai/code/session_01MtdgwpEeCUn6cUA5f1LMPj
This commit is contained in:
Joseph Doherty
2026-07-10 04:14:14 -04:00
parent eb7a913d22
commit 39e1ca3eae
4 changed files with 74 additions and 11 deletions
@@ -73,12 +73,22 @@ public sealed class EmailNotificationDeliveryAdapter : INotificationDeliveryAdap
} }
var smtpConfigs = await _repository.GetAllSmtpConfigurationsAsync(cancellationToken); var smtpConfigs = await _repository.GetAllSmtpConfigurationsAsync(cancellationToken);
var smtpConfig = smtpConfigs.FirstOrDefault(); // Deterministic pick: the lowest-Id row wins regardless of repository order,
// so every node/attempt agrees on the same configuration (arch-review S7).
var smtpConfig = smtpConfigs.OrderBy(c => c.Id).FirstOrDefault();
if (smtpConfig == null) if (smtpConfig == null)
{ {
return DeliveryOutcome.Permanent("No SMTP configuration available"); return DeliveryOutcome.Permanent("No SMTP configuration available");
} }
if (smtpConfigs.Count > 1)
{
_logger.LogWarning(
"Multiple SMTP configurations exist; using the lowest-Id row ({Id}). "
+ "Enforce a single row or delete extras.",
smtpConfig.Id);
}
// An unknown TLS mode is a configuration error that retrying cannot fix — // An unknown TLS mode is a configuration error that retrying cannot fix —
// surface it as a permanent failure (SMTP TLS validation policy). // surface it as a permanent failure (SMTP TLS validation policy).
SmtpTlsMode tlsMode; SmtpTlsMode tlsMode;
@@ -437,7 +437,7 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers
} }
/// <summary> /// <summary>
/// Resolves the SMTP-derived retry policy from the first SMTP configuration row. When no /// Resolves the SMTP-derived retry policy from the lowest-Id SMTP configuration row. When no
/// SMTP configuration exists, falls back to the conservative default constants — delivery /// SMTP configuration exists, falls back to the conservative default constants — delivery
/// itself will permanently fail in that case, so the policy only acts as a guard. /// itself will permanently fail in that case, so the policy only acts as a guard.
/// </summary> /// </summary>
@@ -445,12 +445,22 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers
INotificationRepository notificationRepository) INotificationRepository notificationRepository)
{ {
var configurations = await notificationRepository.GetAllSmtpConfigurationsAsync(); var configurations = await notificationRepository.GetAllSmtpConfigurationsAsync();
var configuration = configurations.Count > 0 ? configurations[0] : null; // Deterministic pick: the lowest-Id row wins regardless of repository order, so the
// retry policy resolves the same way the delivery adapter selects its config (S7).
var configuration = configurations.OrderBy(c => c.Id).FirstOrDefault();
if (configuration is null) if (configuration is null)
{ {
return new RetryPolicy(FallbackMaxRetries, FallbackRetryDelay); return new RetryPolicy(FallbackMaxRetries, FallbackRetryDelay);
} }
if (configurations.Count > 1)
{
_logger.LogWarning(
"Multiple SMTP configurations exist; using the lowest-Id row ({Id}). "
+ "Enforce a single row or delete extras.",
configuration.Id);
}
return ClampRetryPolicy( return ClampRetryPolicy(
nameof(SmtpConfiguration), configuration.MaxRetries, configuration.RetryDelay); nameof(SmtpConfiguration), configuration.MaxRetries, configuration.RetryDelay);
} }
@@ -56,18 +56,22 @@ public class OAuth2TokenService
var key = CredentialKey($"{credentials}|{authority}|{scope}"); var key = CredentialKey($"{credentials}|{authority}|{scope}");
var entry = _cache.GetOrAdd(key, _ => new CacheEntry()); var entry = _cache.GetOrAdd(key, _ => new CacheEntry());
if (entry.Token != null && DateTimeOffset.UtcNow < entry.Expiry) // Read the token/expiry pair through a single volatile reference so a lock-free
// reader can never observe a torn token/expiry combination (arch-review S8).
var current = entry.Current;
if (current != null && DateTimeOffset.UtcNow < current.Expiry)
{ {
return entry.Token; return current.Token;
} }
await entry.Lock.WaitAsync(cancellationToken); await entry.Lock.WaitAsync(cancellationToken);
try try
{ {
// Double-check after acquiring the per-credential lock. // Double-check after acquiring the per-credential lock.
if (entry.Token != null && DateTimeOffset.UtcNow < entry.Expiry) current = entry.Current;
if (current != null && DateTimeOffset.UtcNow < current.Expiry)
{ {
return entry.Token; return current.Token;
} }
var parts = credentials.Split(':', 3); var parts = credentials.Split(':', 3);
@@ -109,8 +113,10 @@ public class OAuth2TokenService
?? throw new InvalidOperationException("No access_token in OAuth2 response"); ?? throw new InvalidOperationException("No access_token in OAuth2 response");
var expiresIn = doc.RootElement.GetProperty("expires_in").GetInt32(); var expiresIn = doc.RootElement.GetProperty("expires_in").GetInt32();
entry.Token = token; // Publish the token/expiry pair as a single atomic reference assignment
entry.Expiry = DateTimeOffset.UtcNow.AddSeconds(expiresIn - 60); // Refresh 60s before expiry // (Refresh 60s before expiry).
entry.Current = new TokenValue(
token, DateTimeOffset.UtcNow.AddSeconds(expiresIn - 60));
// The token endpoint identity is logged by tenant only — never // The token endpoint identity is logged by tenant only — never
// the client secret or the access token itself. // the client secret or the access token itself.
@@ -134,10 +140,14 @@ public class OAuth2TokenService
return Convert.ToHexString(hash); return Convert.ToHexString(hash);
} }
/// <summary>An atomically-published token/expiry pair — one reference, never torn.</summary>
private sealed record TokenValue(string Token, DateTimeOffset Expiry);
private sealed class CacheEntry private sealed class CacheEntry
{ {
public string? Token; // Written only under Lock; read lock-free via the volatile reference so a reader
public DateTimeOffset Expiry = DateTimeOffset.MinValue; // always sees a self-consistent token/expiry pair (or null for "no token yet").
public volatile TokenValue? Current;
public readonly SemaphoreSlim Lock = new(1, 1); public readonly SemaphoreSlim Lock = new(1, 1);
} }
} }
@@ -264,4 +264,37 @@ public class EmailNotificationDeliveryAdapterTests
await Assert.ThrowsAnyAsync<OperationCanceledException>( await Assert.ThrowsAnyAsync<OperationCanceledException>(
() => adapter.DeliverAsync(MakeNotification(), cts.Token)); () => adapter.DeliverAsync(MakeNotification(), cts.Token));
} }
[Fact]
public async Task MultipleSmtpConfigs_LowestIdWins()
{
// Arch-review S7: when more than one SMTP configuration row exists the pick
// must be deterministic (lowest Id), regardless of repository row order.
var list = new NotificationList("ops-team") { Id = 1 };
_repository.GetListByNameAsync("ops-team").Returns(list);
_repository.GetRecipientsByListIdAsync(1).Returns(new List<NotificationRecipient>
{
new("Alice", "alice@example.com") { Id = 1, NotificationListId = 1 }
});
// Repository returns the higher-Id row FIRST — FirstOrDefault would pick it.
_repository.GetAllSmtpConfigurationsAsync().Returns(new List<SmtpConfiguration>
{
new("smtp-seven.example.com", "basic", "noreply@example.com")
{
Id = 7, Port = 587, Credentials = "user:pass", TlsMode = "starttls"
},
new("smtp-three.example.com", "basic", "noreply@example.com")
{
Id = 3, Port = 587, Credentials = "user:pass", TlsMode = "starttls"
}
});
var adapter = CreateAdapter();
await adapter.DeliverAsync(MakeNotification());
// The lowest-Id row (3) must be the one connected to.
await _smtpClient.Received().ConnectAsync(
"smtp-three.example.com", Arg.Any<int>(), Arg.Any<SmtpTlsMode>(),
Arg.Any<int>(), Arg.Any<CancellationToken>());
}
} }