I’m working with a JasperFx Software client who is in the beginning stages of building a pretty complex, multi-step file import process that is going to involve several different services. For the sake of example code in this post, let’s say that we have the (much simplified from my client’s actual logical workflow) workflow from the diagram above:
- External partners (or customers) are sending us an Excel sheet with records that our system will need to process and utilize within our downstream systems (invoices? payments? people? transactions?)
- For the sake of improved throughput, the incoming file is broken into batches of records so the smaller batches can be processed in parallel
- Each batch needs to be validated by the “Validation Service”
- When each batch has been completely validated:
- If there are any errors, send a rejection summary about the entire file to the original external partner
- If there are no errors, try to send each record batch to “Downstream System #1”
- When each batch has been completely accepted or rejected by “Downstream System #1”
- If there are any rejections, send a rejection summary about the entire file to the original external partner
- If all batches are accepted by “Downstream System #1”, try to send each record batch to “Downstream System #2”
- When each batch has been completely accepted or rejected by “Downstream System #2”
- If there are any rejections, send a rejection summary about the entire file to the original external partner and a message to “Downstream System #1” to reverse each previously accepted records in the file
- If all batches are accepted by “Downstream System #2”, send a successful receipt message to the original external partner and archive the intermediate state
Right off the bat, I think we can identify a couple needs and challenges:
- We need some way to track the current, in process state of an individual file and where all the various batches are in that process
- At every point, make decisions about what to do next in the workflow based on the current state of the file based on incremental process. And to make this as clear as possible, I think it’s extremely valuable to be able to clearly write, read, unit test, and reason about this workflow code without any significant coupling to the surrounding infrastructure.
- The whole system should be resilient in the face of the expected transient hiccups like a database getting overwhelmed or a downstream system being temporarily down and “work” should never get lost or hopefully even require human intervention at runtime
- Especially for large files, we absolutely better be prepared for some challenging concurrency issues when lots of incoming messages attempt to update that central file import processing state
- Make it all performance too of course!
Alright, so we’re definitely using both Marten for persistence and Wolverine for the workflow and messaging between services for all of this. The first basic approach for the state management is to use Wolverine’s stateful saga support with Marten. In that case we might have a saga type in Marten something like this:
// Again, express the stages in terms of your
// business domain instead of technical terms,
// but you'll do better than me on this front!
public enum FileImportStage
{
Validating,
Downstream1,
Downstream2,
Completed
}
// As long as it's JSON serialization friendly, you can happily
// tighten up the access here all you want, but I went for quick and simple
public class FileImportSaga :
// Only necessary marker type for Wolverine here
Saga,
// Opts into tracked version concurrency for Marten
// We probably want in this case
IRevisioned
{
// Identity for this saga within our system
public Guid Id { get; set; }
public string FileName { get; set; }
public string PartnerTrackingNumber { get; set; }
public DateTimeOffset Created { get; set; } = DateTimeOffset.UtcNow;
public List<RecordBatchTracker> RecordBatches { get; set; } = new();
public FileImportStage Stage { get; set; } = FileImportStage.Validating;
// Much more in just a bit...
}
Inside our system, we can start a new FileImportSaga and launch the first set of messages to validate each batch of records with this handler that reacts to a request to import a new file:
public record ImportFile(string fileName);
// This could have been done inside the FileImportSaga as well,
// but I think I'd rather keep that focused on the state machine
// and workflow logic
public static class FileImportHandler
{
public static async Task<(FileImportSaga, OutgoingMessages)> Handle(
ImportFile command,
IFileImporter importer,
CancellationToken token)
{
var saga = await importer.ReadAsync(command.fileName, token);
var messages = new OutgoingMessages();
messages.AddRange(saga.CreateValidationMessages());
return (saga, messages);
}
}
public interface IFileImporter
{
Task<FileImportSaga> ReadAsync(string fileName, CancellationToken token);
}
Let’s say that we’re receiving messages back from the Validation Message like this:
public record ValidationResult(Guid Id, Guid BatchId, ValidationMessage[] Messages);
public record ValidationMessage(int RecordNumber, string Message);
Quick note, if Wolverine is handling the messaging in the downstream systems, it’s helping make this easier by tracking the saga id in message metadata from upstream to downstream and back to the upstream through response messages. Otherwise you’d have to track the saga id on the incoming messages.
We could process the validation results in our saga one at a time like so:
// Use Wolverine's cascading message feature here for the next steps
public IEnumerable<object> Handle(ValidationResult validationResult)
{
var currentBatch = RecordBatches
.FirstOrDefault(x => x.Id == validationResult.BatchId);
// We'd probably rig up Wolverine error handling so that it either discards
// a message in this case or immediately moves it to the dead letter queue
// because there's no sense in trying to retry a message that can never be
// processed successfully
if (currentBatch == null) throw new UnknownBatchException(Id, validationResult.BatchId);
currentBatch.ReadValidationResult(validationResult);
var currentValidationStatus = determineValidationStatus();
switch (currentValidationStatus)
{
case RecordStatus.Pending:
yield break;
case RecordStatus.Accepted:
Stage = FileImportStage.Downstream1;
foreach (var batch in RecordBatches)
{
yield return new RequestDownstream1Processing(Id, batch.Id, batch.Records);
}
break;
case RecordStatus.Rejected:
// This saga is complete
MarkCompleted();
// Tell the original sender that this file is rejected
// I'm assuming that Wolverine will get the right information
// back to the original sender somehhow
yield return BuildRejectionMessage();
break;
}
}
private RecordStatus determineValidationStatus()
{
if (RecordBatches.Any(x => x.ValidationStatus == RecordStatus.Pending))
{
return RecordStatus.Pending;
}
if (RecordBatches.Any(x => x.ValidationStatus == RecordStatus.Rejected))
{
return RecordStatus.Rejected;
}
return RecordStatus.Accepted;
}
First off, I’m going to argue that the way that Wolverine supports its stateful sagas and its cascading message feature make the workflow logic pretty easy to unit test in isolation from all the infrastructure. That part is good, right? But what’s maybe not great is that we could easily be getting a bunch of those ValidationResult messages back for the same file at the same time because they’re handled in parallel, so we really need to be prepared for that.
We could rely on the Wolverine/Marten combination’s support for optimistic concurrency and just retry ValidationResult messages that fail because of caught ConcurrencyException, but that’s potentially thrashing the database and the application pretty hard. We could also solve this problem in a “sledgehammer to crack a nut” kind of way by using Wolverine’s strictly ordered listener approach that would force the file import status messages to be processed in order on a single running node:
builder.Host.UseWolverine(opts =>
{
opts.UseRabbitMq(builder.Configuration.GetConnectionString("rabbitmq"));
opts.ListenToRabbitQueue("file-import-updates")
// Single file, serialized access across the
// entire running application cluster!
.ListenWithStrictOrdering();
});
That solves the concurrency issue in a pretty hard core way, but it’s not going to terribly performant because you’ve eliminated all concurrency between different files and you’re making the system constantly load, then save the FileImportSaga data for intermediate steps. Let’s adjust this and incorporate Wolverine’s new message batching feature.
First off, let’s add a new validation batch message like so:
public record ValidationResultBatch(Guid Id, ValidationResult[] Results);
And a new message handler on our saga type for that new message type:
public IEnumerable<object> Handle(ValidationResultBatch batch)
{
var groups = batch.Results.GroupBy(x => x.BatchId);
foreach (var group in groups)
{
var currentBatch = RecordBatches
.FirstOrDefault(x => x.Id == group.Key);
foreach (var result in group)
{
currentBatch.ReadValidationResult(result);
}
}
return DetermineNextStepsAfterValidation();
}
// I pulled this out as a helper, but also, it's something
// you probably want to unit test in isolation on just the FileImportSaga
// class to nail down the workflow logic w/o having to do an integration
// test
public IEnumerable<object> DetermineNextStepsAfterValidation()
{
var currentValidationStatus = determineValidationStatus();
switch (currentValidationStatus)
{
case RecordStatus.Pending:
yield break;
case RecordStatus.Accepted:
Stage = FileImportStage.Downstream1;
foreach (var batch in RecordBatches)
{
yield return new RequestDownstream1Processing(Id, batch.Id, batch.Records);
}
break;
case RecordStatus.Rejected:
// This saga is complete
MarkCompleted();
// Tell the original sender that this file is rejected
// I'm assuming that Wolverine will get the right information
// back to the original sender somehhow
yield return BuildRejectionMessage();
break;
}
}
And lastly, we need to tell Wolverine how to do the message batching, which I’ll do first with this code:
public class ValidationResultBatcher : IMessageBatcher
{
public IEnumerable<Envelope> Group(IReadOnlyList<Envelope> envelopes)
{
var groups = envelopes
.GroupBy(x => x.Message.As<ValidationResult>().Id)
.ToArray();
foreach (var group in groups)
{
var message = new ValidationResultBatch(group.Key, group.OfType<ValidationResult>().ToArray());
// It's important here to pass along the group of envelopes that make up
// this batched message for Wolverine's transactional inbox/outbox
// tracking
yield return new Envelope(message, group);
}
}
public Type BatchMessageType => typeof(ValidationResultBatch);
}
Then lastly, in your Wolverine configuration in your Program file (or a helper method that’s called from Program), you’d tell Wolverine about the batching strategy like so:
builder.Host.UseWolverine(opts =>
{
// Other Wolverine configuration...
opts.BatchMessagesOf<ValidationResult>(x =>
{
x.Batcher = new ValidationResultBatcher();
x.BatchSize = 100;
});
});
With the message batching, you’re potentially putting less load on the database and improving performance by simply making fewer reads and writes over all. You might still have some concurrency concerns, so you have more options to control the parallelization of the ValidationResultBatch messages running locally like this in your UseWolverine() configuration:
opts.LocalQueueFor<ValidationResultBatch>()
// You *could* do this to completely prevent
// concurrency issues
.Sequential()
// Or depend on some level of retries on concurrency
// exceptions and let it parallelize work by file
.MaximumParallelMessages(5);
We could choose to accept some risk of concurrent access to an individual FileImportSaga (unlikely after the batching, but still), so let’s add some better optimistic concurrency checking with our friend Marten. For any given Saga type that’s persisted with Marten, just implement the IRevisioned interface to let Wolverine know to opt into Marten’s concurrency protection like so:
public class FileImportSaga :
// Only necessary marker type for Wolverine here
Saga,
// Opts into tracked version concurrency for Marten
// We probably want in this case
IRevisioned
That’s it, that’s all you need to do. What this does for you is create a check by Wolverine & Marten together that during the processing of any message on a FileImportSaga that no other message was successfully processed against that FileImportSaga between loading the initial copy of the saga at the time the transaction is committed. If Marten detects a concurrency violation upon the commit, it rejects the transaction and throws a ConcurrencyException. We can handle that with a series of retries to just have Wolverine retry the message from the new state with this error handling policy that I’m going to make specific to our FileImportSaga like so:
public class FileImportSaga :
// Only necessary marker type for Wolverine here
Saga,
// Opts into tracked version concurrency for Marten
// We probably want in this case
IRevisioned
{
public static void Configure(HandlerChain chain)
{
// Retry the message over again at least 3 times
// with the specified wait times
chain.OnException<ConcurrencyException>()
.RetryWithCooldown(100.Milliseconds(), 250.Milliseconds(), 250.Milliseconds());
}
// ... the rest of FileImportSaga
See Wolverine’s error handling facilities for more information.
So now we’ve got the beginnings of a multi-step process using Wolverine’s stateful saga support. We’ve also taken some care to protect our file import process against concurrency concerns. And we’ve done all of this in a way where we can quite handily test the workflow logic by just doing state-based tests against the FileImportSaga with no database or message broker infrastructure in sight before we waste any time trying to debug the whole shebang.
Summary
The key takeaway I hope you get from this is that the full Critter Stack has some significant tooling to help you build complex, multi-step workflows. Pair that with the easy getting started stories that both tools have, and I think you have a toolset that allows you to quickly start while also scaling up to more complex needs when you need that.
As so very often happens, this blog post was bigger than I thought it would be, and I’m breaking it up into a series of a follow ups. In the next version of this post, we’ll take the same logical FileImportSaga and do the logical workflow tracking with Marten event sourcing to track the state and use some cool new Marten functionality for the workflow logic inside of Marten projections.
This might take a bit to get to, but I’ll also revisit this original implementation and talk about some extra Marten functionality to further optimize performance by baking in archiving through Marten soft-deletes and its support for PostgreSQL table partitioning.
So historically I’m actually pretty persnickety about being precise about technical terms and design pattern names, but I’m admittedly sloppy about calling something a “Saga” when maybe it’s technically a “Process Manager” and I got jumped online about that by a celebrity programmer. Sorry, not sorry?
The Shade Tree Developer 