Hoe effectief te werken met stateful processen in een stateless omgeving

De service is ook erg simpel: public class PurchaseService { public async Task PurchaseAsync(string orderId) { var order = await _orderGateway.GetOrderAsync(orderId); if (order == null) return; await _orderGateway.CreateInvoiceAsync(orderId); foreach (var product in order.Products) { await _orderGateway.MarkProductAsSoldAsync(product); } } } De gateways in deze voorbeelden zijn klassen die een HttpClient gebruiken om verzoeken naar het externe systeem te verzenden. Ze bevatten de specifieke logica om het Request- en Response-object naar het juiste gegevensformaat te mappen voor elk van de externe API's.

En de testen beginnen...

Toen de eerste implementatie werd getest, werd het duidelijk dat de oorspronkelijke implementatie een beetje te eenvoudig was. Het kon wat optimalisaties gebruiken waarmee we wat repetitieve code konden verwijderen. We besloten om een deel van de code naar verzoekverwerkers te verplaatsen, een concept uit het MediatR-pakket. Dit stelde ons in staat om de directe gateway-aanroepen te vervangen door mediator-aanroepen en het verzoek te bevatten in records die we Commando's en Query's hebben genoemd. Commando's voor acties die dingen doen en Query's voor acties die dingen ophalen. Hoewel het niet specifiek vereist is om deze namen te gebruiken bij het gebruik van MediatR, paste het mooi bij onze manier van werken.



public class MediatorPurchaseService { public async Task PurchaseAsync(string orderId) { var order = await _mediator.Send(new GetOrderQuery(orderId)); await _mediator.Send(new CreateInvoiceCommand(orderId)); foreach (var product in order.Products) { await _mediator.Send(new MarkProductAsSoldCommand(product)); } } }



Toen deze implementatie werd getest, werkte het behoorlijk goed in de testomgeving. Het deed zijn werk, hoewel het een beetje traag was. Niet omdat onze code inefficiënt of slecht was, maar omdat de kwaliteit van de externe API's begon door te schijnen.

Echter, toen de implementatie in de praktijk werd getest, ontdekten we dat het aanroepen van de factuur-aanmaak een achtergrondproces in het facturatiesysteem activeerde dat moest worden afgewacht. Als we doorgaan voordat dat proces is voltooid, lopen we het risico dat bepaalde wijzigingen worden overschreven door het achtergrondproces, waardoor de aankoop in een vreemde, ongedocumenteerde staat achterblijft.

Een alternatieve aanpak

Voordat we terugkeerden naar de tekentafel, bedachten we een snelle oplossing om het probleem op te lossen. We zouden een activiteit gebruiken die werd getriggerd vanuit het oorspronkelijke http-verzoek, en die ons in staat stelde om te wachten op het facturatiesysteem:



[FunctionName("Purchase3")] public async Task<IActionResult> PurchaseAsync( [HttpTrigger(AuthorizationLevel.Function, "post", Route = "example3/purchase/{orderId}")] HttpRequest req, [DurableClient] IDurableOrchestrationClient orchestrationClient, string orderId) { var invoiceId = await _purchaseService.StartPurchaseAsync(orderId); await orchestrationClient.StartNewAsync("Purchase3Orchestrator", new Purchase(orderId, invoiceId)); return new StatusCodeResult(StatusCodes.Status202Accepted); }



En de activity zag er als volgt uit:



[FunctionName("Purchase3Orchestrator")] public async Task OrchestrateAsync([OrchestrationTrigger]IDurableOrchestrationContext context) => await context.CallActivityWithRetryAsync("Purchase3Activity", new RetryOptions(TimeSpan.FromSeconds(1), 10), context.GetInput<Purchase>()); [FunctionName("Purchase3Activity")] public async Task ActivityAsync([ActivityTrigger]Purchase purchase) => await _purchaseService.CompletePurchaseAsync(purchase.OrderId, purchase.InvoiceId);



De purchase call was verdeeld:

public class SplitMediatorPurchaseService { public async Task<Guid> StartPurchaseAsync(string orderId) { var order = await _mediator.Send(new GetOrderQuery(orderId)); var invoice = await _mediator.Send(new CreateInvoiceCommand(order.Id)); return invoice.Id; } public async Task CompletePurchaseAsync(string orderId, Guid invoiceId) { var invoice = await _mediator.Send(new GetInvoiceQuery(invoiceId)); if (invoice.StillProcessing) throw new InvalidOperationException("Invoice is still processing"); var order = await _mediator.Send(new GetOrderQuery(orderId)); foreach (var product in order.Products) { await _mediator.Send(new MarkProductAsSoldCommand(product)); } } }



Toen dit werd ingezet, werkte het prima. Het facturatiesysteem kreeg meer ademruimte en de aankopen werden correct verwerkt. Omdat we een activiteit gebruiken die wordt getriggerd door de runtime en niet door een losse thread die wacht op een Task.Delay, was de oplossing al behoorlijk veerkrachtig tegenover herstarts en nieuwe implementaties. Maar toen we een herhaalpoging activeerden door een uitzondering te gooien, werden onze logboeken overspoeld met foutmeldingen die eigenlijk geen echte fouten waren. Zoals je waarschijnlijk kunt beoordelen aan de hand van de code en de ad-hoc benadering die we in dit geval hebben genomen, was deze oplossing ondermaats en zou het onderhoudsproblemen kunnen veroorzaken.

Terug naar de tekentafel

Bij Triple houden we van een uitdaging. Echter, we houden niet van onderhoudsproblemen veroorzaakt door slecht ontworpen code die snel is ingezet. Dus, welke uitdagingen kan zo'n slecht ontwerp ons opleveren?

Ten eerste is het moeilijk om de losse activiteit te volgen. Het oorspronkelijke http-verzoek activeert een orchestratie die een activiteit activeert. De client ziet geen resultaat van die activiteit, omdat het asynchroon gebeurt. Dus als het mislukt, mislukt het onzichtbaar.

Bovendien wordt het steeds moeilijker om de herbruikbaarheid van de aankoopstroom te behouden. Wat als het proces opnieuw vastloopt en een andere aparte activiteit vereist? Moeten we naast StartPurchase en CompletePurchase ook ContinueWithPurchase maken? En krijgen we dan een ContinueWithPurchase2? Daarnaast wordt de situatie voor unit testing ook ingewikkelder. Als we nog een splitsing moeten maken, moeten we heel wat unit tests aanpassen.

Tot slot heb ik er altijd een hekel aan wanneer het framework te veel de vorm van de oplossing begint te dicteren. Durable Functions, die orchestratie en activiteiten bieden, hebben de neiging om de businesslogica naar de daadwerkelijke functies te trekken. Ik zie Azure Functions (niet alleen activity trigger en orchestrators, maar ook http trigger en service bus triggers) als controllers waarin ik dingen wil valideren en mappen, maar ik zet mijn daadwerkelijke businesslogica graag in aparte services, handlers en dergelijke. Het hebben van een service met toegewijde methoden voor elke activiteitstrigger voelt gewoon als een slechte praktijk.

Orkestratie met Durable Functions

Laten we uitzoomen en beginnen met enkele basisprincipes. Op dat moment was ik nog steeds van plan om deze uitdaging op te lossen met Durable Functions, maar met enkele extra's die ons leven veel gemakkelijker zouden maken.

Laten we beginnen met de orchestraties en activiteiten die Durable Functions ons bieden. Ze zijn krachtig, omdat je orchestrators kunt schrijven die complexe stroomregeling gebruiken, terwijl je de duurzaamheid van automatische herhalingen krijgt. Overweeg de volgende orchestrator:

[FunctionName("Example4")] public async Task OrchestrateAsync([OrchestrationTrigger] IDurableOrchestrationContext context) { try { var data = context.GetInput<Data>(); var productsProcessed = new List<string>(); foreach (var product in data.Products) { await context.CallActivityAsync("ProductActivity", product); productsProcessed.Add(product); } await context.CallActivityAsync("ProductsActivity", productsProcessed); } catch (Exception ex) { await context.CallActivityAsync("FailureActivity", ex); } finally { await context.CallActivityAsync("FinalizingActivity", context.GetInput<string>()); } }



De eerste keer dat deze orchestrator wordt uitgevoerd, komt hij in de foreach-lus en activeert de eerste activiteit. Nadat deze is geactiveerd, stopt de orchestrator en wordt deze na korte tijd opnieuw gestart. De tweede keer dat de orchestrator de activiteit bereikt, controleert deze of de activiteit al is uitgevoerd. Indien wel, gebruikt hij het resultaat van die activiteit en gaat verder. Als die activiteit nog steeds bezig is, stopt de orchestrator en probeert het later opnieuw. Azure Table Storage wordt gebruikt als een opslagplaats om de verzoeken naar en reacties van elke activiteit op te slaan.

De orchestrator speelt alle stappen af ​​die het in eerdere uitvoeringen heeft gedaan en gaat verder waar het gebleven was. De volledige complexe status wordt herhaald en dus hersteld bij elke oproep, waardoor het zeer veerkrachtig is tegenover storingen. Als een orchestratie wordt onderbroken, wordt alles gewoon opnieuw afgespeeld bij de volgende oproep.

Activiteiten zijn functies die worden geactiveerd door de orchestrator. Ze worden alleen uitgevoerd wanneer ze worden geactiveerd en werken op dezelfde manier als http-triggers.

Orchestrators kunnen activiteiten opnieuw proberen waar ze dat nodig achten door ze eenvoudigweg opnieuw aan te roepen, of je kunt aangeven hoe vaak een activiteit opnieuw moet worden geprobeerd voordat deze als mislukt wordt beschouwd. En wanneer een activiteit mislukt, wordt de uitzondering vanuit de activiteit in de orchestrator gegooid, waardoor het behandelen van uitzonderingen als onderdeel van de orchestrator natuurlijk aanvoelt.

Het nadeel van Durable Functions is het feit dat je activiteiten moet oproepen door functies aan te roepen en strings door te geven. Kritieke bedrijfslogica wordt vaak gevuld met activiteitsaanroepen op basis van strings, met dynamische parameters. Het testen van die activiteiten is een pijn. De correlatie tussen activiteiten en orchestrators is enigszins moeilijk te zien, omdat ze alleen op naam gebaseerd zijn, en Visual Studio kan niet echt helpen met zijn refactor-tools wanneer je de handtekening van een activiteit bijwerkt.

Orchestrators en activiteiten zijn echter niet de enige dingen die Durable Functions bieden. Er is ook een functie genaamd Durable Entities, en ze bieden enkele interessante mogelijkheden. Een Durable Entity is een klasse die een bepaalde interface blootstelt, die kan worden aangeroepen vanuit een orchestrator. Beschouw deze interface als een voorbeeld:

public interface IBankAccount { Task AddAsync(decimal amount); Task RemoveAsync(decimal amount); }



Om met deze entiteit te communiceren, moet een orchestrator een proxy naar deze entiteit maken:



[FunctionName("Example5")] public async Task OrchestrateAsync([OrchestrationTrigger]IDurableOrchestrationContext context) { var transaction = context.GetInput<Data>(); var account1 = new EntityId("BankAccount", transaction.From); var account2 = new EntityId("BankAccount", transaction.To); var account1Entity = context.CreateEntityProxy<IBankAccount>(account1); var account2Entity = context.CreateEntityProxy<IBankAccount>(account2); using (await context.LockAsync(account1, account2)) { await account1Entity.RemoveAsync(transaction.Amount); await account2Entity.AddAsync(transaction.Amount); } }



Deze proxy genereert enige IL-code die de benodigde onderdelen configureert om de aanroepargumenten van de orchestrator naar de entiteitstrigger te brengen, die deze doorstuurt naar de juiste entiteit.



public class BankAccount : IBankAccount { public decimal CurrentAmount { get; set; } public Task AddAsync(decimal amount) { CurrentAmount += amount; return Task.CompletedTask; } public Task RemoveAsync(decimal amount) { CurrentAmount -= amount; return Task.CompletedTask; } [FunctionName("BankAccount")] public static Task DispatchAsync([EntityTrigger] IDurableEntityContext context) => context.DispatchAsync<BankAccount>(); }



Als de orchestrator Add aanroept op de proxy, wordt de entiteitstrigger geactiveerd. De entiteitstrigger stuurt het verzoek door naar de entiteit en roept de Add-methode ervan aan. De Durable Entity-abstractie verbergt veel van de complexiteiten van Durable Functions. De orchestrator voert gewoon oproepen uit naar een interface, en vanuit het perspectief van de entiteit wordt de Add-methode eenvoudigweg vanuit een bepaalde context aangeroepen. Het is niet vereist dat de entiteit de interface implementeert, omdat alles nog steeds op strings is gebaseerd. Ik beschouw het echter als goede praktijk om dit toch te doen, omdat de contract, implementatie en het gebruik automatisch op elkaar zijn afgestemd

Orkestreer stappen als workflow

Als we terugkijken naar de implementatie van onze StartPurchase en CompletePurchase, bevatten deze methoden meerdere aanroepen naar de IMediator-interface. Wat zou er gebeuren als we die interface vervangen door een IDurableMediator-interface die een Duurzame Entiteit is die alle triggers doorstuurt naar de echte IMediator? Op deze manier kunnen we de orchestrator implementeren als een reeks mediatoraanroepen. We hebben deze orchestrators 'workflows' genoemd en hebben ervoor gezorgd dat de duurzame mediator een transparante doorvoer is naar elke verzoekverwerker.

public class PurchaseWorkflow { public async Task OrchestrateAsync(IDurableOrchestrationContext context) { var orderId = context.GetInput<string>(); var mediator = context.CreateEntityProxy<IDurableMediator>(new EntityId(nameof(DurableMediator), context.InstanceId)); var order = await mediator.SendAsync(new GetOrderQuery(orderId)); var createdInvoice = await mediator.SendAsync(new CreateInvoiceCommand(order.Id)); do { var invoice = await mediator.SendAsync(new GetInvoiceQuery(createdInvoice.Id)); if (!invoice.StillProcessing) break; await context.CreateTimer(context.CurrentUtcDateTime.AddSeconds(1), CancellationToken.None); } while (true); foreach (var product in order.Products) { await mediator.SendAsync(new MarkProductAsSoldCommand(product)); } } }

Er zijn enkele zeer belangrijke details die worden opgelegd door de Durable Functions-bibliotheek. Generics in activiteiten en entiteiten zijn verboden en activiteiten kunnen slechts één parameter hebben. De verzoekparameters zijn standaard enkelvoudig, maar omdat de commando's worden geserialiseerd en gedeserialiseerd bij het doorgeven van orchestrator naar entiteit, moesten ze worden geserialiseerd met hun typenaam inbegrepen. Dit resulteerde in het verpakken van de verzoeken en reacties in modellen met de juiste JsonProperty-configuratie, die er zo uitzagen:

public class DurableMediator : IDurableMediator { private readonly IMediator _mediator; public DurableMediator(IMediator mediator) { _mediator = mediator; } public async Task SendObjectAsync(WorkflowRequest request) { await _mediator.Send(request.Request); } public async Task<WorkflowResponse> SendObjectWithResponseAsync(WorkflowRequestWithResponse request) { // the dynamic is needed for the dynamic dispatch of Send() var result = await _mediator.Send(request.Request); return new WorkflowResponse(result); } }

[JsonObject(ItemTypeNameHandling = TypeNameHandling.All)] public record WorkflowRequest(IRequest<Unit> Request); [JsonObject(ItemTypeNameHandling = TypeNameHandling.All)] public record WorkflowRequestWithResponse(dynamic Request); [JsonObject(ItemTypeNameHandling = TypeNameHandling.All)] public record WorkflowResponse(object Response); public static class DurableMediatorExtensions { public static async Task<TResponse> SendAsync<TResponse>(this IDurableMediator mediator, IRequest<TResponse> request) { if (typeof(TResponse) == typeof(Unit)) { await mediator.SendObjectAsync(new WorkflowRequest((IRequest<Unit>)request)); return default!; } var response = await mediator.SendObjectWithResponseAsync(new WorkflowRequestWithResponse((IRequest<object>)request)); if (response == null) throw new InvalidOperationException("Received an empty response"); return (TResponse)response.Response; } }



Omdat de aankoopworkflow nu wordt vastgelegd in een aparte klasse die verschillende interfaces gebruikt en unit tests schrijft die het daadwerkelijke gedrag van de andere systemen valideren en simuleren, konden we enkele van de uitzonderlijke gedragingen volledig testen die we tegenkwamen bij het werken met de echte systemen. Merk op dat de wachtfase in de oorspronkelijke implementatie die ons dwong om methoden op te splitsen en meer activiteitstriggers toe te voegen, nu is teruggebracht tot een kleine implementatiedetail in de workflow.

Het enige dat nog ontbreekt, is een orchestrator die de workflow uitvoert. De service kan worden bijgewerkt om een ​​aanroep van de aankoopworkflow aan te vragen.



public class WorkflowPurchaseService { private readonly IDurableClient _durableClient; public WorkflowPurchaseService(IDurableClient durableClient) { _durableClient = durableClient; } public async Task<string> PurchaseAsync(string orderId) => await _durableClient.StartNewAsync("PurchaseWorkflowOrchestrator", input: orderId); }



Om dit voorbeeld eenvoudig te houden, heb ik de details weggelaten over hoe we voorkomen dat we de naam van de orchestrator daar hardcoderen; we hebben dit op een vergelijkbare manier opgelost als hoe MediatR handlers oplost voor elk van de verzoeken.

Terugkijkend heeft deze aanpak ons enkele voordelen opgeleverd die de observeerbaarheid van deze workflows aanzienlijk hebben vergroot en enkele leuke extra's hebben mogelijk gemaakt. De workflow- en mediatorverzoeken worden niet beïnvloed door de structuur die Durable Functions normaal gesproken opleggen. De mediatorverzoeken kunnen gemakkelijk worden hergebruikt, terwijl de workflows volledig kunnen worden getest. Het is gemakkelijk om te zien welke verzoeken door welke workflows worden gebruikt.

Het implementeren van herhalingen of vertragingen is gratis. Het is gemakkelijk om een ​​vertraging van 10 minuten in een workflow te implementeren en te weten dat de workflow daarna wordt hervat. Wachten op een specifieke toestand is eenvoudig, omdat je dezelfde API-aanroep kunt voortzetten en blijven wachten als deze nog niet beschikbaar is.

We hebben ook enkele belangrijke extra's toegevoegd om het bijhouden van lopende workflows te vergemakkelijken. Ten eerste hebben we enkele trace-identificatoren toegevoegd aan de workflow en activiteiten. Alle logs van workflows en activiteiten hebben dezelfde trace-identificatie, waardoor het eenvoudig is om meerdere logs te correleren met één workflow-uitvoering. Daarnaast kan de status van de orchestraties worden waargenomen door andere API's. Wanneer een aankoopworkflow wordt uitgevoerd, kunnen we weergeven dat de overeenkomstige bestelling in het proces van aankoop is of een tweede aankoopverzoek blokkeren wanneer de eerste nog wordt verwerkt.

In onze monitoring nemen we de status op van elke uitgevoerde workflow en tonen we een degradatie van de gezondheid wanneer workflows niet worden voltooid. Omdat de exacte inhoud van de verzoeken en reacties van en naar de activiteiten wordt opgeslagen in de geschiedenisopslag, kunnen we gedetailleerde debuginformatie weergeven voor workflows die zijn mislukt.

Nadelen en oplossingen

Deze aanpak heeft enkele nadelen die we moeten aanpakken. Ten eerste zijn deze Durable Entities niet echt bedoeld om op deze manier te worden gebruikt. Ze hebben persistente staat, en elke aanroep naar deze entiteiten resulteert in het opslaan van enige staat in de tabelopslag, terwijl de duurzame mediator geen enkele staat bevat. Aan de andere kant slaan ze de exacte inhoud van het verzoek en de reacties naar en van de duurzame mediator op, waardoor het gemakkelijker wordt om te debuggen na een mislukte workflow. Als je dit implementeert met activiteiten, verlies je de invoergegevens.

Bovendien zijn sommige herhaalfuncties niet beschikbaar bij het werken met entiteiten. Naast CallActivity heb je CallActivityWithRetry, waarmee automatische herhalingen van activiteiten die mogelijk kunnen mislukken mogelijk zijn. Aangezien je rechtstreeks met de (geproxyde) interface van een entiteit communiceert, moet je de herhaallogica zelf implementeren. Dit is meestal geen probleem, tenzij je ook gebruik wilt maken van de terugspoelfunctie van Durable Functions. Hoewel deze functie nog in de previewfase is en enkele problemen heeft met de automatische herhalingen (https://github.com/Azure/durabletask/issues/811), is het goed om oplossingen te verkennen die niet afhankelijk zijn van Durable Entities.

Een manier om dit op te lossen is opnieuw gebruik te maken van activiteiten op basis van strings, maar het triggeren van die activiteiten te verbergen achter een interface. In plaats van IDurableMediator als de daadwerkelijke entiteitsproxy te hebben, kan het ook een eenvoudige klasse zijn die de juiste activiteit aanroept bij het ontvangen van een Send-oproep. Dit verwijdert de behoefte aan een Durable Entity en vereenvoudigt het gehele opzet van het workflowsysteem. Ik heb deze benadering verkend, inclusief de mogelijkheid om mislukte workflows terug te spoelen, en het gepubliceerd als een NuGet-pakket (https://github.com/ThomasBleijendaal/DurableMediator), dat hier te vinden is. Je kunt daar ook een voorbeeld vinden dat laat zien hoe je een workflow kunt testen.



Conclusie

De workflow-benadering heeft ons een oplossing geboden voor meerdere problemen waar we tegenaan liepen. Ten eerste gaf het ons een duidelijk patroon voor het uitvoeren van langzame of foutgevoelige processen die niet binnen de levensduur van een enkele HTTP API-oproep kunnen worden afgehandeld. Omdat elke ontwikkelaar in het team begrijpt wat een workflow inhoudt, wordt bij het schatten en verfijnen van taken duidelijk hoe iets moet worden opgebouwd met het idee van 'plaats dat proces in een workflow'.

De goede observeerbaarheid van deze workflows maakt het mogelijk om ze te gebruiken voor veel verschillende processen, en we passen dit patroon vrij snel toe. Onze gezondheidsprobes detecteren automatisch alle workflows die zijn mislukt, waardoor ze zeer zichtbaar zijn. We hebben een portal ontwikkeld waarmee we een volledige workflow-run kunnen diagnosticeren, inclusief het verzoek en de reacties van elke activiteit.

Aangezien de implementatie van de activiteiten eenvoudige mediator-verzoekhandlers omvat, is de overstap naar (of terugkeer van) workflows klein en is het gemakkelijk om iets om te zetten in een workflow wanneer iets is ontwikkeld met behulp van verzoekhandlers.

Wat betreft de oorspronkelijke aankoopworkflow, deze is aanzienlijk geëvolueerd. Het maken van de factuur bleek nog complexer te zijn, wat de oprichting van een speciale 'create-invoice'-workflow rechtvaardigde. Deze workflow wordt gestart door de aankoopworkflow als een sub-orchestratie, maar wordt niet afgewacht en duurt langer dan de aankoopworkflow. Alle API-oproepen naar de externe service zijn omhuld met herhalingslogica die het oproepen opnieuw probeert na het tijdelijk onderbreken van de workflow gedurende een korte minuut. De aankoopworkflow zelf is nu vrij eenvoudig en voltooit vrij snel, de meeste complexiteiten worden omgeleid naar de 'create invoice'-workflow. Naast deze twee workflows hebben we workflows gebouwd voor het annuleren van aankopen, andere soorten aankopen en abonnementsverlengingen. We hebben ook unit tests gebouwd die de workflows blootstellen aan zeer onbetrouwbare API-mock-ups, waarbij we hun reactie op mislukking bij elke stap in het proces testen.

Alle codevoorbeelden zijn beschikbaar als werkende voorbeelden in deze repository. Al met al was het een interessante oefening die ons veel heeft geleerd. We kijken ernaar uit om onze ervaring toe te passen op nog complexere projecten!