Merge pull request #242 from simplify9/hamza/fix/retry-total-attempts-cap

Enforce MaxAttemptsTotal across messages instead of per message

Author: hamzahalq

SW.Bitween.Api/Data/BitweenDbContext.cs

@@ -236,10 +236,17 @@ protected override void OnModelCreating(ModelBuilder modelBuilder)
                 b.HasKey(p => p.Id);
                 b.Property(p => p.Id).IsUnicode(false).HasMaxLength(50);
                 b.Property(p => p.On);
-                b.Property(p => p.GroupAttemptCounts).StoreAsJson();
                 b.HasIndex(p => p.On);
             });
 
+            modelBuilder.Entity<RetryGroupUsage>(b =>
+            {
+                b.ToTable("RetryGroupUsages");
+                b.HasKey(p => new { p.SubscriptionId, p.GroupId });
+                b.Property(p => p.AttemptsUsed);
+                b.Property(p => p.LastAttemptOn);
+            });
+
             modelBuilder.Entity<Xchange>(b =>
             {
                 b.ToTable("Xchanges");
@@ -252,7 +259,6 @@ protected override void OnModelCreating(ModelBuilder modelBuilder)
                 b.Property(p => p.HandlerId).HasMaxLength(200).IsUnicode(false);
                 b.Property(p => p.HandlerProperties).StoreAsJson();
                 b.Property(p => p.MapperProperties).StoreAsJson();
-                b.Property(p => p.GroupAttemptCounts).StoreAsJson();
                 b.Property(p => p.InputContentType).IsUnicode(false).HasMaxLength(200);
                 b.Property(p => p.ResponseMessageTypeName).IsUnicode(false).HasMaxLength(500);
 
@@ -283,6 +289,7 @@ protected override void OnModelCreating(ModelBuilder modelBuilder)
                 b.Property(p => p.ResponseName).HasMaxLength(200);
                 b.Property(p => p.ResponseContentType).IsUnicode(false).HasMaxLength(200);
                 b.Property(p => p.OutputContentType).IsUnicode(false).HasMaxLength(200);
+                b.Property(p => p.RetryBlockedReason).HasMaxLength(500);
 
 
                 b.HasOne<Xchange>().WithOne().HasForeignKey<XchangeResult>(p => p.Id).OnDelete(DeleteBehavior.Cascade);

SW.Bitween.Api/Domain/DelayedRetry.cs

@@ -1,11 +1,9 @@
 using System;
-using System.Collections.Generic;
 using SW.PrimitiveTypes;
 
 namespace SW.Bitween.Domain;
 // Id should be the same for xchangeId when retry happens the record is deleted
 public class DelayedRetry : BaseEntity<string>
 {
     public DateTime On { get; set; }
-    public Dictionary<string, int> GroupAttemptCounts { get; set; } = new();
 }

SW.Bitween.Api/Domain/RetryGroupUsage.cs

@@ -0,0 +1,27 @@
+using System;
+
+namespace SW.Bitween.Domain;
+
+/// <summary>
+/// Running total of the retries one retry group has spent for one integration, backing
+/// <c>RetryBudget.MaxAttemptsTotal</c>.  That cap is shared by every message hitting the
+/// group, so it cannot be tracked on an individual xchange.
+/// </summary>
+/// <remarks>
+/// The total never resets on its own: once <see cref="AttemptsUsed"/> reaches the group's
+/// <c>MaxAttemptsTotal</c> the group stops retrying for that integration until this row is
+/// cleared.
+/// </remarks>
+public class RetryGroupUsage
+{
+    /// <summary>The integration whose budget this is. A shared policy gives each one its own total.</summary>
+    public int SubscriptionId { get; set; }
+
+    /// <summary><c>RetryGroup.Id</c>, which survives policy edits, so the total does too.</summary>
+    public Guid GroupId { get; set; }
+
+    public int AttemptsUsed { get; set; }
+
+    /// <summary>When the last attempt was claimed — the only clue left once a group is exhausted.</summary>
+    public DateTime LastAttemptOn { get; set; }
+}

SW.Bitween.Api/Domain/Xchange/Xchange.cs

@@ -60,7 +60,7 @@ public Xchange(Subscription subscription, XchangeFile file, string[] references
         }
 
         //retry xchange
-        public Xchange(Xchange xchange, XchangeFile file, IWorkGroup workGroup, IReadOnlyDictionary<string, int> groupAttemptCounts = null) :
+        public Xchange(Xchange xchange, XchangeFile file, IWorkGroup workGroup) :
             this(xchange.DocumentId, workGroup, file, xchange.References)
         {
             SubscriptionId = xchange.SubscriptionId;
@@ -72,11 +72,10 @@ public Xchange(Xchange xchange, XchangeFile file, IWorkGroup workGroup, IReadOnl
             ResponseSubscriptionId = xchange.ResponseSubscriptionId;
             RetryFor = xchange.Id;
             CorrelationId = xchange.CorrelationId;
-            GroupAttemptCounts = groupAttemptCounts == null ? null : new Dictionary<string, int>(groupAttemptCounts);
         }
 
         //retry with reset subscription properties
-        public Xchange(Subscription subscription, Xchange xchange, XchangeFile file, IReadOnlyDictionary<string, int> groupAttemptCounts = null) :
+        public Xchange(Subscription subscription, Xchange xchange, XchangeFile file) :
             this(xchange.DocumentId, subscription.WorkGroup, file, xchange.References)
         {
             SubscriptionId = xchange.SubscriptionId;
@@ -88,7 +87,6 @@ public Xchange(Subscription subscription, Xchange xchange, XchangeFile file, IRe
             ResponseSubscriptionId = subscription.ResponseSubscriptionId;
             RetryFor = xchange.Id;
             CorrelationId = xchange.CorrelationId;
-            GroupAttemptCounts = groupAttemptCounts == null ? null : new Dictionary<string, int>(groupAttemptCounts);
         }
 
         public int? SubscriptionId { get; private set; }
@@ -109,6 +107,5 @@ public Xchange(Subscription subscription, Xchange xchange, XchangeFile file, IRe
 
         public string RetryFor { get; private set; }
         public string CorrelationId { get; set; }
-        public IReadOnlyDictionary<string, int> GroupAttemptCounts { get; private set; }
     }
 }

SW.Bitween.Api/Domain/XchangeResult/XchangeResult.cs

@@ -62,6 +62,16 @@ public XchangeResult(string xchangeId,WorkGroup workGroup, XchangeFile outputFil
         public bool ResponseBad { get; private set; }
         public string ResponseContentType { get; private set; }
 
+        /// <summary>
+        /// Why the retry policy declined to schedule another attempt for this failure, or
+        /// <c>null</c> when a retry was scheduled or no policy applied. Without it a group that
+        /// has exhausted its budget looks identical to one that never matched.
+        /// </summary>
+        public string RetryBlockedReason { get; private set; }
+
+        /// <summary>Records the policy's refusal so it can be shown alongside the failure.</summary>
+        public void SetRetryBlocked(string reason) => RetryBlockedReason = reason;
+
 
 
     }

SW.Bitween.Api/Resources/RetryPolicies/Delete.cs

@@ -28,7 +28,18 @@ public async Task<object> Handle(int key)
         if (inUse)
             throw new SWException("Cannot delete a retry policy that is assigned to one or more subscriptions.");
 
+        // Same reason as Update: the policy's groups are about to stop existing, so clear their
+        // usage rows rather than strand them.
+        var policy = await _dbContext.FindAsync<RetryPolicy>(key);
+        var groupIds = policy.Groups.Select(g => g.Id).ToList();
+
         await _dbContext.DeleteByKeyAsync<RetryPolicy>(key);
+
+        if (groupIds.Count > 0)
+            await _dbContext.Set<RetryGroupUsage>()
+                .Where(u => groupIds.Contains(u.GroupId))
+                .ExecuteDeleteAsync();
+
         return null;
     }
 }

SW.Bitween.Api/Resources/RetryPolicies/ResetUsage.cs

@@ -0,0 +1,56 @@
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.EntityFrameworkCore;
+using SW.Bitween.Domain;
+using SW.Bitween.Domain.Accounts;
+using SW.Bitween.Model;
+using SW.PrimitiveTypes;
+
+namespace SW.Bitween.Resources.RetryPolicies;
+
+/// <summary>
+/// Clears spent group budget, letting an exhausted group retry again. The total never resets on
+/// its own, so this is the only way back for an integration that has hit its ceiling.
+/// </summary>
+[HandlerName("resetusage")]
+public class ResetUsage : ICommandHandler<int, RetryPolicyResetUsage, object>
+{
+    private readonly BitweenDbContext _dbContext;
+    private readonly RequestContext _requestContext;
+
+    public ResetUsage(BitweenDbContext dbContext, RequestContext requestContext)
+    {
+        _dbContext = dbContext;
+        _requestContext = requestContext;
+    }
+
+    public async Task<object> Handle(int key, RetryPolicyResetUsage request)
+    {
+        _requestContext.EnsureAccess(AccountRole.Admin, AccountRole.Member);
+
+        var policy = await _dbContext.Set<RetryPolicy>().AsNoTracking()
+            .FirstOrDefaultAsync(p => p.Id == key);
+        if (policy == null) throw new SWNotFoundException(key.ToString());
+
+        // Scope the reset to this policy's own integrations and groups, so a policy id in the
+        // route can never clear a counter belonging to a different policy.
+        var subscriptionIds = await _dbContext.Set<Subscription>()
+            .Where(s => s.RetryPolicyId == key)
+            .Select(s => s.Id)
+            .ToListAsync();
+
+        var groupIds = policy.Groups.Select(g => g.Id).ToList();
+
+        var query = _dbContext.Set<RetryGroupUsage>()
+            .Where(u => subscriptionIds.Contains(u.SubscriptionId) && groupIds.Contains(u.GroupId));
+
+        if (request.SubscriptionId.HasValue)
+            query = query.Where(u => u.SubscriptionId == request.SubscriptionId.Value);
+
+        if (request.GroupId.HasValue)
+            query = query.Where(u => u.GroupId == request.GroupId.Value);
+
+        await query.ExecuteDeleteAsync();
+        return null;
+    }
+}

SW.Bitween.Api/Resources/RetryPolicies/Test.cs

@@ -21,7 +21,7 @@ public Test(RequestContext requestContext)
         _requestContext = requestContext;
     }
 
-    public Task<object> Handle(TestRetryPolicyRequest request)
+    public async Task<object> Handle(TestRetryPolicyRequest request)
     {
         _requestContext.EnsureAccess(AccountRole.Admin, AccountRole.Member);
 
@@ -30,13 +30,14 @@ public Task<object> Handle(TestRetryPolicyRequest request)
                 "Choose Error or Bad result — a successful result is never retried.");
 
         var policy = new CustomRetryPolicy { Groups = request.Groups ?? [] };
-        var evaluator = new RetryPolicyEvaluator(policy);
+        // In-memory budget: a dry-run must not spend any real integration's total.
+        var evaluator = new RetryPolicyEvaluator(policy, new InMemoryRetryGroupBudget());
         var attemptsToSimulate = Math.Clamp(request.AttemptsToSimulate, 1, 20);
 
         var attempts = new List<TestRetryAttemptResult>();
         for (var attemptIndex = 0; attemptIndex < attemptsToSimulate; attemptIndex++)
         {
-            var decision = evaluator.Evaluate(request.ResultType, request.Content, attemptIndex);
+            var decision = await evaluator.Evaluate(request.ResultType, request.Content, attemptIndex);
 
             attempts.Add(new TestRetryAttemptResult
             {
@@ -53,6 +54,6 @@ public Task<object> Handle(TestRetryPolicyRequest request)
             if (!decision.ShouldRetry) break;
         }
 
-        return Task.FromResult<object>(new TestRetryPolicyResponse { Attempts = attempts });
+        return new TestRetryPolicyResponse { Attempts = attempts };
     }
 }

SW.Bitween.Api/Resources/RetryPolicies/Update.cs

@@ -1,4 +1,6 @@
+using System.Linq;
 using System.Threading.Tasks;
+using Microsoft.EntityFrameworkCore;
 using SW.Bitween.Domain;
 using SW.Bitween.Domain.Accounts;
 using SW.Bitween.Model;
@@ -23,9 +25,23 @@ public async Task<object> Handle(int key, RetryPolicyUpdate model)
         RetryGroupValidation.EnsureCanFire(model.Groups);
 
         var entity = await _dbContext.FindAsync<RetryPolicy>(key);
+
+        // Spent budget is keyed by group id, so a group removed here would leave usage rows
+        // that no policy claims — invisible to the usage report and beyond the reach of reset.
+        var removedGroupIds = entity.Groups
+            .Select(g => g.Id)
+            .Except((model.Groups ?? []).Select(g => g.Id))
+            .ToList();
+
         entity.Name = model.Name;
         entity.Groups = model.Groups ?? [];
         await _dbContext.SaveChangesAsync();
+
+        if (removedGroupIds.Count > 0)
+            await _dbContext.Set<RetryGroupUsage>()
+                .Where(u => removedGroupIds.Contains(u.GroupId))
+                .ExecuteDeleteAsync();
+
         return null;
     }
 }

SW.Bitween.Api/Resources/RetryPolicies/Usage.cs

@@ -0,0 +1,74 @@
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.EntityFrameworkCore;
+using SW.Bitween.Domain;
+using SW.Bitween.Domain.Accounts;
+using SW.Bitween.Model;
+using SW.PrimitiveTypes;
+
+namespace SW.Bitween.Resources.RetryPolicies;
+
+/// <summary>
+/// Reports how much of each group's total budget the integrations using this policy have spent,
+/// so an exhausted group is visible instead of just silently declining to retry.
+/// </summary>
+[HandlerName("usage")]
+public class Usage : ICommandHandler<int, RetryPolicyUsageRequest, object>
+{
+    private readonly BitweenDbContext _dbContext;
+    private readonly RequestContext _requestContext;
+
+    public Usage(BitweenDbContext dbContext, RequestContext requestContext)
+    {
+        _dbContext = dbContext;
+        _requestContext = requestContext;
+    }
+
+    public async Task<object> Handle(int key, RetryPolicyUsageRequest request)
+    {
+        _requestContext.EnsureAccess(AccountRole.Admin, AccountRole.Member);
+
+        var policy = await _dbContext.Set<RetryPolicy>().AsNoTracking()
+            .FirstOrDefaultAsync(p => p.Id == key);
+        if (policy == null) throw new SWNotFoundException(key.ToString());
+
+        var subscriptions = await _dbContext.Set<Subscription>().AsNoTracking()
+            .Where(s => s.RetryPolicyId == key)
+            .Select(s => new { s.Id, s.Name })
+            .ToListAsync();
+
+        var subscriptionIds = subscriptions.Select(s => s.Id).ToList();
+
+        var usages = await _dbContext.Set<RetryGroupUsage>().AsNoTracking()
+            .Where(u => subscriptionIds.Contains(u.SubscriptionId))
+            .ToListAsync();
+
+        // Only groups that allow retries have a budget to spend.
+        var budgets = policy.Groups
+            .Where(g => g.Budget != null)
+            .ToDictionary(g => g.Id, g => new { g.Name, g.Budget.MaxAttemptsTotal });
+
+        var names = subscriptions.ToDictionary(s => s.Id, s => s.Name);
+
+        var rows = usages
+            .Where(u => budgets.ContainsKey(u.GroupId))
+            .Select(u => new RetryGroupUsageRow
+            {
+                SubscriptionId = u.SubscriptionId,
+                SubscriptionName = names.GetValueOrDefault(u.SubscriptionId),
+                GroupId = u.GroupId,
+                GroupName = budgets[u.GroupId].Name,
+                AttemptsUsed = u.AttemptsUsed,
+                MaxAttemptsTotal = budgets[u.GroupId].MaxAttemptsTotal,
+                Exhausted = u.AttemptsUsed >= budgets[u.GroupId].MaxAttemptsTotal,
+                LastAttemptOn = u.LastAttemptOn
+            })
+            // Exhausted integrations first — those are the ones no longer being retried.
+            .OrderByDescending(r => r.Exhausted)
+            .ThenByDescending(r => r.AttemptsUsed)
+            .ToList();
+
+        return new List<RetryGroupUsageRow>(rows);
+    }
+}