From 3cf499e5161a9b5998f728fe47ab10a14abc1dc6 Mon Sep 17 00:00:00 2001 From: John Simons Date: Tue, 4 Aug 2026 12:57:33 +1000 Subject: [PATCH 1/2] Implement the retry consumer on EF Core The batch lifecycle maps to statements rather than a session: staging is a join from claims to messages, handing a batch to the forwarder is one transaction of updates and deletes plus the pointer row, and the message count is a COUNT over claims. --- .../FailedErrorImportDataStore.cs | 4 +- .../Implementation/FailedMessageViewMapper.cs | 4 +- .../Implementation/RetryStagingStore.cs | 225 ++++++++++++++++-- .../EFRecoverabilityIngestionUnitOfWork.cs | 2 +- .../UnitOfWork/HeadersJsonContext.cs | 8 - .../Infrastructure/MessageHeaders.cs | 18 ++ ...ontrol.Persistence.Tests.PostgreSql.csproj | 1 - ...Control.Persistence.Tests.SqlServer.csproj | 1 - .../EFCore/PersistenceTestsContext.cs | 4 +- 9 files changed, 236 insertions(+), 31 deletions(-) delete mode 100644 src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/HeadersJsonContext.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Infrastructure/MessageHeaders.cs diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs index 3c0d9f6065..d5fd7d7a70 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs @@ -43,7 +43,7 @@ public Task StoreFailedErrorImport(FailedErrorImport failure) => } var failedAt = timeProvider.GetUtcNow().UtcDateTime; - var headersJson = JsonSerializer.Serialize(failure.Message.Headers, HeadersJsonContext.Default.DictionaryStringString); + var headersJson = MessageHeaders.Write(failure.Message.Headers); byte[] storedBody = storeExternally ? [] : body; await dbContext.UpsertAsync([uniqueMessageId], () => new FailedErrorImportEntity @@ -143,7 +143,7 @@ async Task> ReadBatch(int offset, CancellationToke async Task ToTransportMessage(FailedErrorImportEntity import, CancellationToken cancellationToken) { - var headers = JsonSerializer.Deserialize(import.HeadersJson, HeadersJsonContext.Default.DictionaryStringString) ?? []; + var headers = MessageHeaders.Read(import.HeadersJson); var body = import.BodyStoredExternally ? await ReadExternalBody(import.UniqueMessageId, cancellationToken) diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageViewMapper.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageViewMapper.cs index d9115160ca..211151c0ee 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageViewMapper.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageViewMapper.cs @@ -6,6 +6,7 @@ namespace ServiceControl.Persistence.EFCore.Implementation; using ServiceControl.MessageFailures.Api; using ServiceControl.Operations; using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Infrastructure; using ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; static class FailedMessageViewMapper @@ -147,6 +148,5 @@ static ExceptionDetails ToExceptionDetails(this FailedMessageEntity entity, Dict HostId = entity.ReceivingEndpointHostId ?? Guid.Empty }; - static Dictionary ReadHeaders(this FailedMessageEntity entity) => - JsonSerializer.Deserialize(entity.HeadersJson, HeadersJsonContext.Default.DictionaryStringString) ?? []; + static Dictionary ReadHeaders(this FailedMessageEntity entity) => MessageHeaders.Read(entity.HeadersJson); } diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/RetryStagingStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/RetryStagingStore.cs index 0cfe6c9c76..aa2f992134 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/RetryStagingStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/RetryStagingStore.cs @@ -1,34 +1,231 @@ namespace ServiceControl.Persistence.EFCore.Implementation; -public class RetryStagingStore : IRetryStagingStore +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using ServiceControl.MessageFailures; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Infrastructure; + +public class RetryStagingStore(IServiceScopeFactory scopeFactory, TimeProvider timeProvider) : DataStoreBase(scopeFactory), IRetryStagingStore { public Task GetStagingBatch() => - throw new NotImplementedException(); + ExecuteWithDbContext(async dbContext => + { + var batch = await dbContext.RetryBatches + .AsNoTracking() + .Where(batch => batch.Status == RetryBatchStatus.Staging) + .OrderBy(batch => batch.StartTime) + .ThenBy(batch => batch.Id) + .FirstOrDefaultAsync(); + + return batch?.ToRetryBatch(await CountMessages(dbContext, batch.Id)); + }); public Task GetMessagesToStage(string batchId) => - throw new NotImplementedException(); + ExecuteWithDbContext(async dbContext => + { + var batch = ParseBatchId(batchId); + + // A message claimed by an earlier batch is not claimed by this one, and a claim whose + // message is gone drops out of the join, which is what leaves it out of the staging. + var rows = await dbContext.FailedMessageRetries + .AsNoTracking() + .Where(retry => retry.RetryBatchId == batch) + .Join(dbContext.FailedMessages.AsNoTracking(), + retry => retry.UniqueMessageId, + message => message.UniqueMessageId, + (retry, message) => new + { + message.UniqueMessageId, + message.MessageId, + message.FailingEndpointAddress, + message.HeadersJson, + retry.StageAttempts + }) + .ToListAsync(); + + return rows.Select(row => new StagingMessage( + row.UniqueMessageId.ToString(), + row.UniqueMessageId.ToString(), + row.MessageId!, + row.FailingEndpointAddress, + MessageHeaders.Read(row.HeadersJson), + row.StageAttempts)).ToArray(); + }); public Task MarkBatchAsForwarding(string batchId, string stagingId, IReadOnlyCollection stagedMessageIds) => - throw new NotImplementedException(); + ExecuteWithDbContext(async dbContext => + { + var batch = ParseBatchId(batchId); + var staged = ParseMessageIds(stagedMessageIds); + var now = timeProvider.GetUtcNow().UtcDateTime; + + await InTransaction(dbContext, async () => + { + await dbContext.RetryBatches + .Where(row => row.Id == batch) + .ExecuteUpdateAsync(setters => setters + .SetProperty(row => row.Status, RetryBatchStatus.Forwarding) + .SetProperty(row => row.StagingId, stagingId)); + + // The batch keeps only what it staged, so its message count is what the forwarder is + // told to expect. The claims dropped here are of messages that no longer exist. + await dbContext.FailedMessageRetries + .Where(row => row.RetryBatchId == batch && !staged.Contains(row.UniqueMessageId)) + .ExecuteDeleteAsync(); + + await dbContext.FailedMessages + .Where(row => staged.Contains(row.UniqueMessageId)) + .ExecuteUpdateAsync(setters => setters + .SetProperty(row => row.Status, FailedMessageStatus.RetryIssued) + .SetProperty(row => row.StatusChangedAt, now) + .SetProperty(row => row.LastModified, now)); + + await PointForwarderAt(dbContext, batch); + }); + }); public Task DiscardBatch(string batchId) => - throw new NotImplementedException(); + ExecuteWithDbContext(async dbContext => + { + var batch = ParseBatchId(batchId); + + await InTransaction(dbContext, async () => + { + // Nothing was staged, so every claim of this batch is of a message that is gone. + await dbContext.FailedMessageRetries + .Where(row => row.RetryBatchId == batch) + .ExecuteDeleteAsync(); + + await dbContext.RetryBatches + .Where(row => row.Id == batch) + .ExecuteDeleteAsync(); + }); + }); public Task GetForwardingBatchId() => - throw new NotImplementedException(); + ExecuteWithDbContext(async dbContext => + { + var nowForwarding = await dbContext.RetryBatchNowForwarding + .AsNoTracking() + .SingleOrDefaultAsync(); + + return nowForwarding?.RetryBatchId.ToString(); + }); public Task GetBatch(string batchId, CancellationToken cancellationToken) => - throw new NotImplementedException(); + ExecuteWithDbContext(async dbContext => + { + var batch = ParseBatchId(batchId); + + var entity = await dbContext.RetryBatches + .AsNoTracking() + .SingleOrDefaultAsync(row => row.Id == batch, cancellationToken); + + return entity?.ToRetryBatch(await CountMessages(dbContext, batch, cancellationToken)); + }); public Task CompleteForwarding(string batchId) => - throw new NotImplementedException(); + ExecuteWithDbContext(async dbContext => + { + var batch = ParseBatchId(batchId); + + await InTransaction(dbContext, async () => + { + // The claims outlive the batch: they are what stops a message being staged again + // before its retry is confirmed. + await dbContext.RetryBatches + .Where(row => row.Id == batch) + .ExecuteDeleteAsync(); + + await dbContext.RetryBatchNowForwarding + .Where(row => row.RetryBatchId == batch) + .ExecuteDeleteAsync(); + }); + }); + + public Task RecordStagingFailure(IReadOnlyCollection uniqueMessageIds) + { + var failed = ParseMessageIds(uniqueMessageIds); + + return ExecuteWithDbContext(dbContext => dbContext.FailedMessageRetries + .Where(row => failed.Contains(row.UniqueMessageId)) + .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.StageAttempts, 1))); + } + + public Task IncrementStagingAttempts(string uniqueMessageId) + { + if (!Guid.TryParse(uniqueMessageId, out var message)) + { + return Task.CompletedTask; + } + + return ExecuteWithDbContext(dbContext => dbContext.FailedMessageRetries + .Where(row => row.UniqueMessageId == message) + .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.StageAttempts, row => row.StageAttempts + 1))); + } + + public Task RemoveFromBatch(string uniqueMessageId) + { + if (!Guid.TryParse(uniqueMessageId, out var message)) + { + return Task.CompletedTask; + } + + return ExecuteWithDbContext(dbContext => dbContext.FailedMessageRetries + .Where(row => row.UniqueMessageId == message) + .ExecuteDeleteAsync()); + } + + static async Task PointForwarderAt(ServiceControlDbContext dbContext, Guid batch) + { + var updated = await dbContext.RetryBatchNowForwarding + .Where(row => row.Id == RetryBatchNowForwardingEntity.SingleRowId) + .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.RetryBatchId, batch)); + + if (updated == 0) + { + dbContext.RetryBatchNowForwarding.Add(new RetryBatchNowForwardingEntity { RetryBatchId = batch }); + await dbContext.SaveChangesAsync(); + } + } + + static Task CountMessages(ServiceControlDbContext dbContext, Guid batch, CancellationToken cancellationToken = default) => + dbContext.FailedMessageRetries + .AsNoTracking() + .CountAsync(retry => retry.RetryBatchId == batch, cancellationToken); + + static Task InTransaction(ServiceControlDbContext dbContext, Func operations) => + dbContext.Database.CreateExecutionStrategy().ExecuteAsync(async () => + { + await using var transaction = await dbContext.Database.BeginTransactionAsync(); + + await operations(); + + await transaction.CommitAsync(); + }); + + // Message ids reach this from the API, so an id that is not a message id cannot match a stored + // message and is left out rather than thrown at. + static HashSet ParseMessageIds(IReadOnlyCollection uniqueMessageIds) + { + var parsed = new HashSet(uniqueMessageIds.Count); - public Task RecordStagingFailure(IReadOnlyCollection uniqueMessageIds) => - throw new NotImplementedException(); + foreach (var uniqueMessageId in uniqueMessageIds) + { + if (Guid.TryParse(uniqueMessageId, out var message)) + { + parsed.Add(message); + } + } - public Task IncrementStagingAttempts(string uniqueMessageId) => - throw new NotImplementedException(); + return parsed; + } - public Task RemoveFromBatch(string uniqueMessageId) => - throw new NotImplementedException(); + // Batch ids only ever come from CreateBatch, so anything else is a programming error. + static Guid ParseBatchId(string batchId) => + Guid.TryParse(batchId, out var parsed) + ? parsed + : throw new ArgumentException($"'{batchId}' is not a retry batch id issued by this store.", nameof(batchId)); } diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs index 92d3a37ab5..4a2067264b 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs @@ -36,7 +36,7 @@ public Task RecordFailedProcessingAttempt(MessageContext context, AttemptedAt = processingAttempt.AttemptedAt, TimeOfFailure = processingAttempt.FailureDetails.TimeOfFailure, Groups = groups, - HeadersJson = JsonSerializer.Serialize(processingAttempt.Headers, HeadersJsonContext.Default.DictionaryStringString), + HeadersJson = MessageHeaders.Write(processingAttempt.Headers), MessageId = processingAttempt.MessageId, MessageType = GetMetadata(processingAttempt, "MessageType"), TimeSent = GetMetadata(processingAttempt, "TimeSent"), diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/HeadersJsonContext.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/HeadersJsonContext.cs deleted file mode 100644 index c10617e8fe..0000000000 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/HeadersJsonContext.cs +++ /dev/null @@ -1,8 +0,0 @@ -namespace ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; - -using System.Text.Json.Serialization; - -// Source generated serialization for the failed message's headers, which are stored verbatim as -// the HeadersJson column. Avoids the reflection-based serializer on the ingestion hot path. -[JsonSerializable(typeof(Dictionary))] -partial class HeadersJsonContext : JsonSerializerContext; diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/MessageHeaders.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/MessageHeaders.cs new file mode 100644 index 0000000000..c8c66d83c2 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/MessageHeaders.cs @@ -0,0 +1,18 @@ +namespace ServiceControl.Persistence.EFCore.Infrastructure; + +using System.Text.Json; +using System.Text.Json.Serialization; + +// The headers of a failed message are stored verbatim as the HeadersJson column. +static class MessageHeaders +{ + public static string Write(Dictionary headers) => + JsonSerializer.Serialize(headers, HeadersJsonContext.Default.DictionaryStringString); + + public static Dictionary Read(string headersJson) => + JsonSerializer.Deserialize(headersJson, HeadersJsonContext.Default.DictionaryStringString) ?? []; +} + +// Source generated serialization, which keeps the reflection-based serializer off the ingestion hot path. +[JsonSerializable(typeof(Dictionary))] +partial class HeadersJsonContext : JsonSerializerContext; diff --git a/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj b/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj index 6452921127..a8cd23fb3c 100644 --- a/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj +++ b/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj @@ -35,7 +35,6 @@ - diff --git a/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj b/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj index 3e50556865..af49f36dcd 100644 --- a/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj +++ b/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj @@ -35,7 +35,6 @@ - diff --git a/src/ServiceControl.Persistence.Tests/EFCore/PersistenceTestsContext.cs b/src/ServiceControl.Persistence.Tests/EFCore/PersistenceTestsContext.cs index 7476d806bd..f7b0de3b11 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/PersistenceTestsContext.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/PersistenceTestsContext.cs @@ -8,7 +8,7 @@ namespace ServiceControl.Persistence.Tests; using System.Threading.Tasks; using EFCore.DbContexts; using EFCore.Entities; -using EFCore.Implementation.UnitOfWork; +using EFCore.Infrastructure; using MessageFailures; using Microsoft.Extensions.DependencyInjection; using NServiceBus; @@ -41,7 +41,7 @@ static async Task InsertFailedMessagesDirect(IServiceProvider serviceProvider, F FirstTimeOfFailure = ordered.Min(pa => pa.FailureDetails.TimeOfFailure), LastTimeOfFailure = ordered.Max(pa => pa.FailureDetails.TimeOfFailure), LastAttemptedAt = attempt.AttemptedAt, - HeadersJson = JsonSerializer.Serialize(attempt.Headers, HeadersJsonContext.Default.DictionaryStringString), + HeadersJson = MessageHeaders.Write(attempt.Headers), MessageId = attempt.MessageId, MessageType = GetMetadata(attempt, "MessageType"), TimeSent = GetMetadata(attempt, "TimeSent"), From 462e7ec3626facf39cd2ea9aa9872d70728f9b44 Mon Sep 17 00:00:00 2001 From: John Simons Date: Tue, 4 Aug 2026 16:59:27 +1000 Subject: [PATCH 2/2] Addressed feedback --- .../Implementation/RetryStagingStore.cs | 76 ++++++++++--------- 1 file changed, 42 insertions(+), 34 deletions(-) diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/RetryStagingStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/RetryStagingStore.cs index aa2f992134..467871666e 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/RetryStagingStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/RetryStagingStore.cs @@ -22,11 +22,12 @@ public class RetryStagingStore(IServiceScopeFactory scopeFactory, TimeProvider t return batch?.ToRetryBatch(await CountMessages(dbContext, batch.Id)); }); - public Task GetMessagesToStage(string batchId) => - ExecuteWithDbContext(async dbContext => - { - var batch = ParseBatchId(batchId); + public Task GetMessagesToStage(string batchId) + { + var batch = ParseBatchId(batchId); + return ExecuteWithDbContext(async dbContext => + { // A message claimed by an earlier batch is not claimed by this one, and a claim whose // message is gone drops out of the join, which is what leaves it out of the staging. var rows = await dbContext.FailedMessageRetries @@ -53,15 +54,16 @@ public Task GetMessagesToStage(string batchId) => MessageHeaders.Read(row.HeadersJson), row.StageAttempts)).ToArray(); }); + } - public Task MarkBatchAsForwarding(string batchId, string stagingId, IReadOnlyCollection stagedMessageIds) => - ExecuteWithDbContext(async dbContext => - { - var batch = ParseBatchId(batchId); - var staged = ParseMessageIds(stagedMessageIds); - var now = timeProvider.GetUtcNow().UtcDateTime; + public Task MarkBatchAsForwarding(string batchId, string stagingId, IReadOnlyCollection stagedMessageIds) + { + var batch = ParseBatchId(batchId); + var staged = ParseMessageIds(stagedMessageIds); + var now = timeProvider.GetUtcNow().UtcDateTime; - await InTransaction(dbContext, async () => + return ExecuteWithDbContext(dbContext => + InTransaction(dbContext, async () => { await dbContext.RetryBatches .Where(row => row.Id == batch) @@ -83,15 +85,15 @@ await dbContext.FailedMessages .SetProperty(row => row.LastModified, now)); await PointForwarderAt(dbContext, batch); - }); - }); + })); + } - public Task DiscardBatch(string batchId) => - ExecuteWithDbContext(async dbContext => - { - var batch = ParseBatchId(batchId); + public Task DiscardBatch(string batchId) + { + var batch = ParseBatchId(batchId); - await InTransaction(dbContext, async () => + return ExecuteWithDbContext(dbContext => + InTransaction(dbContext, async () => { // Nothing was staged, so every claim of this batch is of a message that is gone. await dbContext.FailedMessageRetries @@ -101,8 +103,8 @@ await dbContext.FailedMessageRetries await dbContext.RetryBatches .Where(row => row.Id == batch) .ExecuteDeleteAsync(); - }); - }); + })); + } public Task GetForwardingBatchId() => ExecuteWithDbContext(async dbContext => @@ -114,24 +116,26 @@ await dbContext.RetryBatches return nowForwarding?.RetryBatchId.ToString(); }); - public Task GetBatch(string batchId, CancellationToken cancellationToken) => - ExecuteWithDbContext(async dbContext => - { - var batch = ParseBatchId(batchId); + public Task GetBatch(string batchId, CancellationToken cancellationToken) + { + var batch = ParseBatchId(batchId); + return ExecuteWithDbContext(async dbContext => + { var entity = await dbContext.RetryBatches .AsNoTracking() .SingleOrDefaultAsync(row => row.Id == batch, cancellationToken); return entity?.ToRetryBatch(await CountMessages(dbContext, batch, cancellationToken)); }); + } - public Task CompleteForwarding(string batchId) => - ExecuteWithDbContext(async dbContext => - { - var batch = ParseBatchId(batchId); + public Task CompleteForwarding(string batchId) + { + var batch = ParseBatchId(batchId); - await InTransaction(dbContext, async () => + return ExecuteWithDbContext(dbContext => + InTransaction(dbContext, async () => { // The claims outlive the batch: they are what stops a message being staged again // before its retry is confirmed. @@ -142,8 +146,8 @@ await dbContext.RetryBatches await dbContext.RetryBatchNowForwarding .Where(row => row.RetryBatchId == batch) .ExecuteDeleteAsync(); - }); - }); + })); + } public Task RecordStagingFailure(IReadOnlyCollection uniqueMessageIds) { @@ -206,8 +210,10 @@ static Task InTransaction(ServiceControlDbContext dbContext, Func operatio await transaction.CommitAsync(); }); - // Message ids reach this from the API, so an id that is not a message id cannot match a stored - // message and is left out rather than thrown at. + /// + /// Message ids reach this from the API, so an id that is not a message id cannot match a stored + /// message and is left out rather than thrown at. + /// static HashSet ParseMessageIds(IReadOnlyCollection uniqueMessageIds) { var parsed = new HashSet(uniqueMessageIds.Count); @@ -223,7 +229,9 @@ static HashSet ParseMessageIds(IReadOnlyCollection uniqueMessageId return parsed; } - // Batch ids only ever come from CreateBatch, so anything else is a programming error. + /// + /// Batch ids only ever come from CreateBatch, so anything else is a programming error. + /// static Guid ParseBatchId(string batchId) => Guid.TryParse(batchId, out var parsed) ? parsed