From 8b27e7b7d1d6d7372dcf9273725629912da26c5d Mon Sep 17 00:00:00 2001 From: Brian Tyler Date: Fri, 24 Jul 2026 12:22:55 +0100 Subject: [PATCH 1/2] feat: add smooth GCRA rate limits for concurrency groups --- README.md | 13 +- .../LauncherPageRenderer.cs | 1 + samples/Sheddueller.SampleHost/Program.cs | 26 ++ samples/Sheddueller.SampleHost/README.md | 1 + .../Components/Pages/ConcurrencyGroups.razor | 383 +++++++++++++++++- .../Components/_Imports.razor | 1 + .../Internal/DashboardFilters.cs | 8 +- ...earConcurrencyDefaultRateLimitOperation.cs | 16 + ...arConcurrencyRateLimitOverrideOperation.cs | 20 + ...gresConcurrencyGroupInspectionOperation.cs | 135 +++++- .../PostgresConcurrencyRateLimits.cs | 146 +++++++ .../Internal/Operations/PostgresJobGroups.cs | 18 +- ...SetConcurrencyDefaultRateLimitOperation.cs | 16 + .../SetConcurrencyRateLimitOperation.cs | 20 + ...tConcurrencyUnlimitedRateLimitOperation.cs | 20 + .../Operations/TryClaimNextJobOperation.cs | 39 +- .../Internal/PostgresJobStore.cs | 50 +++ .../Internal/PostgresMigrator.cs | 56 +++ .../Internal/PostgresNames.cs | 2 +- .../Internal/ShedduellerWorker.cs | 39 +- src/Sheddueller/ConcurrencyGroupRateLimit.cs | 10 + .../ConcurrencyGroupRateLimitOverride.cs | 10 + .../ConcurrencyGroupRateLimitOverrideKind.cs | 22 + src/Sheddueller/IConcurrencyGroupManager.cs | 50 +++ .../ConcurrencyGroupInspectionDetail.cs | 13 +- .../ConcurrencyGroupInspectionQuery.cs | 3 +- .../ConcurrencyGroupInspectionSummary.cs | 40 ++ .../Logging/ShedduellerLoggerMessages.cs | 44 ++ .../Runtime/ConcurrencyGroupManager.cs | 115 ++++++ src/Sheddueller/Storage/ClaimJobResult.cs | 3 +- ...ClearConcurrencyDefaultRateLimitRequest.cs | 8 + ...learConcurrencyRateLimitOverrideRequest.cs | 8 + src/Sheddueller/Storage/IJobStore.cs | 48 +++ .../SetConcurrencyDefaultRateLimitRequest.cs | 9 + .../Storage/SetConcurrencyRateLimitRequest.cs | 9 + ...SetConcurrencyUnlimitedRateLimitRequest.cs | 8 + .../DashboardEndpointTests.cs | 22 +- .../DashboardFilterTests.cs | 2 + .../SetConcurrencyRateLimitOperationTests.cs | 88 ++++ .../TryClaimNextJobOperationTests.cs | 138 +++++++ .../PostgresMigrationTests.cs | 33 ++ .../PostgresTestContext.cs | 37 +- .../InspectionContractTests.cs | 54 +++ .../ConcurrencyGroupManagerTests.cs | 83 ++++ test/Sheddueller.Tests/RecordingJobStore.cs | 65 +++ .../WorkerRateLimitTimingTests.cs | 42 ++ 46 files changed, 1927 insertions(+), 47 deletions(-) create mode 100644 src/Sheddueller.Postgres/Internal/Operations/ClearConcurrencyDefaultRateLimitOperation.cs create mode 100644 src/Sheddueller.Postgres/Internal/Operations/ClearConcurrencyRateLimitOverrideOperation.cs create mode 100644 src/Sheddueller.Postgres/Internal/Operations/PostgresConcurrencyRateLimits.cs create mode 100644 src/Sheddueller.Postgres/Internal/Operations/SetConcurrencyDefaultRateLimitOperation.cs create mode 100644 src/Sheddueller.Postgres/Internal/Operations/SetConcurrencyRateLimitOperation.cs create mode 100644 src/Sheddueller.Postgres/Internal/Operations/SetConcurrencyUnlimitedRateLimitOperation.cs create mode 100644 src/Sheddueller/ConcurrencyGroupRateLimit.cs create mode 100644 src/Sheddueller/ConcurrencyGroupRateLimitOverride.cs create mode 100644 src/Sheddueller/ConcurrencyGroupRateLimitOverrideKind.cs create mode 100644 src/Sheddueller/Storage/ClearConcurrencyDefaultRateLimitRequest.cs create mode 100644 src/Sheddueller/Storage/ClearConcurrencyRateLimitOverrideRequest.cs create mode 100644 src/Sheddueller/Storage/SetConcurrencyDefaultRateLimitRequest.cs create mode 100644 src/Sheddueller/Storage/SetConcurrencyRateLimitRequest.cs create mode 100644 src/Sheddueller/Storage/SetConcurrencyUnlimitedRateLimitRequest.cs create mode 100644 test/Sheddueller.Postgres.Tests/Operations/SetConcurrencyRateLimitOperationTests.cs create mode 100644 test/Sheddueller.Worker.Tests/WorkerRateLimitTimingTests.cs diff --git a/README.md b/README.md index 9ee23f3..af2b37a 100644 --- a/README.md +++ b/README.md @@ -120,7 +120,18 @@ Use `UsePostgres(postgres => postgres.DataSource = dataSource)` when an applicat The operational store keeps active jobs plus a bounded searchable terminal window. By default, background retention cleanup keeps completed jobs for 24 hours and failed or canceled jobs for 7 days, then deletes those terminal job rows and their tags, concurrency groups, and events. Configure `ShedduellerOptions.JobRetention` to change the windows, set a state retention to `null` to keep that state indefinitely, or set `Enabled = false` to disable cleanup. -Concurrency group limits use a persisted override over a code-defined default over the built-in default of `1`. Use `IConcurrencyGroupManager.SetDefaultLimitAsync(...)` from startup or deployment seeding code so dashboard edits survive restarts. Use `SetLimitAsync(...)` for an explicit live override and `ClearLimitOverrideAsync(...)` to fall back to the code default. +Concurrency groups independently enforce active-job capacity and an optional smooth job-start rate across the cluster. Capacity limits use a persisted override over a code-defined default over the built-in default of `1`. Rate limits use a persisted override over a code-defined default over a built-in unlimited rate. + +Use `IConcurrencyGroupManager.SetDefaultLimitAsync(...)` and `SetDefaultRateLimitAsync(...)` from startup or deployment seeding code so dashboard edits survive restarts. Rate permits are evenly spaced: a rate of two starts per second admits one claim every 500 milliseconds and does not accumulate burst credit while idle. Every successful claim consumes a rate permit, including retries and reclaims. + +```csharp +await concurrencyGroups.SetDefaultLimitAsync("provider:happy-holiday-homes", 1); +await concurrencyGroups.SetDefaultRateLimitAsync( + "provider:happy-holiday-homes", + new ConcurrencyGroupRateLimit(2, TimeSpan.FromSeconds(1))); +``` + +Use `SetLimitAsync(...)` and `SetRateLimitAsync(...)` for limited live overrides. `SetUnlimitedRateLimitAsync(...)` explicitly disables a code-defined rate, while `ClearRateLimitOverrideAsync(...)` returns to the code default. `ClearLimitOverrideAsync(...)` performs the equivalent reset for capacity. ## Enqueue Jobs diff --git a/samples/Sheddueller.SampleHost/LauncherPageRenderer.cs b/samples/Sheddueller.SampleHost/LauncherPageRenderer.cs index b685af7..10488f7 100644 --- a/samples/Sheddueller.SampleHost/LauncherPageRenderer.cs +++ b/samples/Sheddueller.SampleHost/LauncherPageRenderer.cs @@ -73,6 +73,7 @@ public static string Render(string? statusMessage) AppendActionCard(builder, "/launch/delayed", "Delayed job", "Queues a short delayed job to exercise delayed state and not-before time.", "Enqueue job"); AppendActionCard(builder, "/launch/many-tags", "Many tags", "Queues a tagged job with informational tags first and ceremonial tags later.", "Enqueue job"); AppendActionCard(builder, "/launch/blocking-batch", "Concurrency batch", "Sets a shared group limit to 1 and enqueues several long jobs.", "Enqueue batch"); + AppendActionCard(builder, "/launch/rate-limited-batch", "Rate-limited batch", "Queues six jobs with concurrency 3 and a smooth rate of two starts per five seconds.", "Enqueue batch"); AppendActionCard(builder, "/launch/idempotent", "Idempotent reprice", "Queues one reprice-listing-3 job behind a held group slot; click twice quickly to reuse the queued job.", "Enqueue job"); AppendActionCard(builder, "/launch/cancelable", "Cancelable delayed job", "Creates a delayed queued job that can be canceled from the dashboard.", "Enqueue job"); builder.AppendLine(" "); diff --git a/samples/Sheddueller.SampleHost/Program.cs b/samples/Sheddueller.SampleHost/Program.cs index 745f8a5..7c50a72 100644 --- a/samples/Sheddueller.SampleHost/Program.cs +++ b/samples/Sheddueller.SampleHost/Program.cs @@ -138,6 +138,32 @@ return RedirectWithMessage($"Queued {jobIds.Count} concurrency-demo jobs in group '{GroupKey}' with limit 1."); }); +app.MapPost("/launch/rate-limited-batch", async ( + IConcurrencyGroupManager concurrencyGroupManager, + IJobEnqueuer enqueuer, + CancellationToken cancellationToken) => +{ + const string GroupKey = "demo:rate-limited"; + await concurrencyGroupManager.SetDefaultLimitAsync(GroupKey, 3, cancellationToken).ConfigureAwait(false); + await concurrencyGroupManager.SetDefaultRateLimitAsync( + GroupKey, + new ConcurrencyGroupRateLimit(2, TimeSpan.FromSeconds(5)), + cancellationToken).ConfigureAwait(false); + + var jobIds = new List(); + for (var index = 1; index <= 6; index++) + { + var jobId = await enqueuer.EnqueueAsync( + (service, ct) => service.RunQuickAsync($"rate-limited-{index}", ct), + new JobSubmission(Priority: 20, ConcurrencyGroupKeys: [GroupKey]), + cancellationToken).ConfigureAwait(false); + jobIds.Add(jobId); + } + + return RedirectWithMessage( + $"Queued {jobIds.Count} jobs in group '{GroupKey}' with concurrency 3 and a smooth rate of 2 starts per 5 seconds."); +}); + app.MapPost("/launch/idempotent", async ( IConcurrencyGroupManager concurrencyGroupManager, IJobEnqueuer enqueuer, diff --git a/samples/Sheddueller.SampleHost/README.md b/samples/Sheddueller.SampleHost/README.md index 57b6af9..bb4ddd6 100644 --- a/samples/Sheddueller.SampleHost/README.md +++ b/samples/Sheddueller.SampleHost/README.md @@ -52,6 +52,7 @@ The sample applies PostgreSQL schema migrations automatically on startup and reg - `Permanent failure`: terminal failure without retries - `Delayed job`: waits 30 seconds before becoming claimable - `Concurrency batch`: sets a shared limit of 1 and queues several long-running jobs +- `Rate-limited batch`: queues six jobs with concurrency 3 and a smooth rate of two starts every five seconds - `Idempotent reprice`: queues a 10-second reprice job with generated idempotency behind a group limit of 1; click twice quickly to see the same queued job reused - `Recurring demo`: creates or updates a recurring schedule that fires each minute - `Cancelable delayed job`: creates a queued delayed job that can be canceled from the dashboard job detail page diff --git a/src/Sheddueller.Dashboard/Components/Pages/ConcurrencyGroups.razor b/src/Sheddueller.Dashboard/Components/Pages/ConcurrencyGroups.razor index c2c7964..5e00805 100644 --- a/src/Sheddueller.Dashboard/Components/Pages/ConcurrencyGroups.razor +++ b/src/Sheddueller.Dashboard/Components/Pages/ConcurrencyGroups.razor @@ -52,6 +52,7 @@ + @@ -61,6 +62,9 @@ + + + @@ -75,6 +79,9 @@ Group Key Effective Limit Limit Source + Effective Rate + Rate Source + Next Permit Current Occupancy Blocked Jobs Saturation State @@ -114,6 +121,46 @@ @LimitSourceText(group) + + @if (IsEditingRate(group)) + { +
+ + per + + +
+ } + else + { + @FormatRateLimit(group.EffectiveRateLimit) + } + + + @RateLimitSourceText(group) + + + @if (group.EffectiveRateLimit is null) + { + Unlimited + } + else if (!group.IsRateLimited) + { + Available now + } + else + { + + } + @DashboardFormat.Count(group.CurrentOccupancy) @DashboardFormat.Count(group.BlockedJobCount) @@ -141,6 +188,17 @@ } + else if (IsEditingRate(group)) + { + + + } else { + + + } @@ -238,7 +310,7 @@ } .groups-table { - min-width: 1320px; + min-width: 1840px; table-layout: fixed; white-space: nowrap; } @@ -255,6 +327,10 @@ width: 148px; } + .groups-table__col--rate { + width: 300px; + } + .groups-table__col--state { width: 176px; } @@ -264,7 +340,7 @@ } .groups-table__col--actions { - width: 116px; + width: 232px; } .groups-table thead { @@ -306,6 +382,10 @@ background: color-mix(in srgb, var(--sd-error) 8%, transparent); } + .groups-row--rate-limited { + background: color-mix(in srgb, var(--sd-warning) 10%, transparent); + } + .groups-table__number { text-align: right; } @@ -313,6 +393,7 @@ .groups-key, .groups-count, .groups-blocked-count, + .groups-rate-available, .groups-timestamp__absolute { font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", "Courier New", monospace; } @@ -331,6 +412,24 @@ font-weight: 700; } + .groups-rate-editor { + display: flex; + align-items: center; + gap: 4px; + } + + .groups-rate-count-input { + width: 60px; + } + + .groups-rate-period-input { + width: 76px; + } + + .groups-rate-unit-input { + width: 88px; + } + .groups-state { display: inline-flex; align-items: center; @@ -497,6 +596,9 @@ @code { private static readonly TimeSpan PageRefreshInterval = TimeSpan.FromSeconds(5); private const int DefaultPageSize = 25; + private const string SecondsUnit = "seconds"; + private const string MinutesUnit = "minutes"; + private const string HoursUnit = "hours"; private readonly List _groups = []; private readonly DashboardConcurrencyGroupFilters _filters = new(); @@ -505,7 +607,11 @@ private string? _loadError; private string? _actionMessage; private string? _editingGroupKey; + private string? _editingRateGroupKey; private int _limitInput = 1; + private int _ratePermitInput = 1; + private decimal _ratePeriodInput = 1; + private string _ratePeriodUnit = SecondsUnit; private bool _isLoading; private bool _isLoadingMore; private bool _isActionRunning; @@ -518,7 +624,7 @@ => this._manager is not null; private int GroupTableColumnCount - => this.CanEditLimits ? 8 : 7; + => this.CanEditLimits ? 11 : 10; private string ActionAlertClass => this._isActionError @@ -584,6 +690,15 @@ await this.LoadAsync(); } + private async Task SetRateLimitedOnlyAsync(bool value) + { + this._filters.RateLimitedOnly = value; + this.ClearActionAlert(); + this.CancelEdit(); + + await this.LoadAsync(); + } + private void SetGroupKeyFilter(string value) { this._filters.GroupKey = value; @@ -662,6 +777,11 @@ { this.CancelEdit(); } + + if (this._editingRateGroupKey is not null && page.Groups.All(group => !string.Equals(group.GroupKey, this._editingRateGroupKey, StringComparison.Ordinal))) + { + this.CancelEdit(); + } } private void StartEdit(ConcurrencyGroupInspectionSummary group) @@ -672,12 +792,31 @@ } this.ClearActionAlert(); + this._editingRateGroupKey = null; this._editingGroupKey = group.GroupKey; this._limitInput = group.OverrideLimit ?? group.EffectiveLimit; } private void CancelEdit() - => this._editingGroupKey = null; + { + this._editingGroupKey = null; + this._editingRateGroupKey = null; + } + + private void StartRateEdit(ConcurrencyGroupInspectionSummary group) + { + if (this.IsActionDisabled) + { + return; + } + + this.ClearActionAlert(); + this._editingGroupKey = null; + this._editingRateGroupKey = group.GroupKey; + var rateLimit = group.OverrideRateLimit ?? group.EffectiveRateLimit ?? new ConcurrencyGroupRateLimit(1, TimeSpan.FromSeconds(1)); + this._ratePermitInput = rateLimit.PermitCount; + (this._ratePeriodInput, this._ratePeriodUnit) = ToRatePeriodInput(rateLimit.Period); + } private async Task SaveLimitAsync(ConcurrencyGroupInspectionSummary group) { @@ -743,9 +882,105 @@ } } + private async Task SaveRateLimitAsync(ConcurrencyGroupInspectionSummary group) + { + if (this._manager is null || this.IsActionDisabled || !this.IsEditingRate(group)) + { + return; + } + + if (!this.TryCreateRateLimit(out var rateLimit, out var validationMessage)) + { + this.SetActionFailure(string.Concat("Rate-limit save failed: ", validationMessage)); + return; + } + + this._isActionRunning = true; + this.ClearActionAlert(); + + try + { + await this._manager.SetRateLimitAsync(group.GroupKey, rateLimit); + this.CancelEdit(); + this.SetActionSuccess(string.Create( + CultureInfo.InvariantCulture, + $"Concurrency group {group.GroupKey} rate override set to {FormatRateLimit(rateLimit)}.")); + await this.LiveRefresh.RefreshNowAsync(); + } + catch (Exception exception) + { + this.SetActionFailure(string.Create(CultureInfo.InvariantCulture, $"Rate-limit save failed: {exception.Message}")); + } + finally + { + this._isActionRunning = false; + } + } + + private async Task SetUnlimitedRateLimitAsync(ConcurrencyGroupInspectionSummary group) + { + if (this._manager is null || this.IsActionDisabled) + { + return; + } + + this._isActionRunning = true; + this.ClearActionAlert(); + + try + { + await this._manager.SetUnlimitedRateLimitAsync(group.GroupKey); + this.CancelEdit(); + this.SetActionSuccess(string.Create( + CultureInfo.InvariantCulture, + $"Concurrency group {group.GroupKey} rate override set to unlimited.")); + await this.LiveRefresh.RefreshNowAsync(); + } + catch (Exception exception) + { + this.SetActionFailure(string.Create(CultureInfo.InvariantCulture, $"Unlimited rate override failed: {exception.Message}")); + } + finally + { + this._isActionRunning = false; + } + } + + private async Task ResetRateLimitAsync(ConcurrencyGroupInspectionSummary group) + { + if (this._manager is null || this.IsActionDisabled) + { + return; + } + + this._isActionRunning = true; + this.ClearActionAlert(); + + try + { + await this._manager.ClearRateLimitOverrideAsync(group.GroupKey); + this.CancelEdit(); + this.SetActionSuccess(string.Create( + CultureInfo.InvariantCulture, + $"Concurrency group {group.GroupKey} rate override reset.")); + await this.LiveRefresh.RefreshNowAsync(); + } + catch (Exception exception) + { + this.SetActionFailure(string.Create(CultureInfo.InvariantCulture, $"Rate-limit reset failed: {exception.Message}")); + } + finally + { + this._isActionRunning = false; + } + } + private bool IsEditing(ConcurrencyGroupInspectionSummary group) => string.Equals(this._editingGroupKey, group.GroupKey, StringComparison.Ordinal); + private bool IsEditingRate(ConcurrencyGroupInspectionSummary group) + => string.Equals(this._editingRateGroupKey, group.GroupKey, StringComparison.Ordinal); + private void ClearActionAlert() { this._actionMessage = null; @@ -765,7 +1000,9 @@ } private static string GroupRowClass(ConcurrencyGroupInspectionSummary group) - => group.IsSaturated ? "groups-row--saturated" : string.Empty; + => group.IsSaturated + ? "groups-row--saturated" + : group.IsRateLimited ? "groups-row--rate-limited" : string.Empty; private static string BlockedCountClass(ConcurrencyGroupInspectionSummary group) => group.BlockedJobCount > 0 ? "groups-blocked-count groups-blocked-count--attention" : "groups-blocked-count"; @@ -777,6 +1014,11 @@ return "Saturated"; } + if (group.IsRateLimited) + { + return "Rate Limited"; + } + if (group.BlockedJobCount > 0) { return "Blocked Work"; @@ -789,11 +1031,13 @@ { var modifier = group.IsSaturated ? "saturated" - : group.BlockedJobCount > 0 + : group.IsRateLimited ? "blocked" - : GetOccupancyRatio(group) >= 0.8 - ? "high" - : "nominal"; + : group.BlockedJobCount > 0 + ? "blocked" + : GetOccupancyRatio(group) >= 0.8 + ? "high" + : "nominal"; return string.Concat("groups-state groups-state--", modifier); } @@ -834,4 +1078,125 @@ private static string ResetLimitLabel(ConcurrencyGroupInspectionSummary group) => string.Create(CultureInfo.InvariantCulture, $"Reset limit override for concurrency group {group.GroupKey}"); + + private static string FormatRateLimit(ConcurrencyGroupRateLimit? rateLimit) + => rateLimit is null + ? "Unlimited" + : string.Create( + CultureInfo.InvariantCulture, + $"{DashboardFormat.Count(rateLimit.PermitCount)} / {FormatRatePeriod(rateLimit.Period)}"); + + private static string FormatRatePeriod(TimeSpan period) + { + if (period.Ticks % TimeSpan.TicksPerHour == 0) + { + return string.Create(CultureInfo.InvariantCulture, $"{period.TotalHours:0.###} h"); + } + + if (period.Ticks % TimeSpan.TicksPerMinute == 0) + { + return string.Create(CultureInfo.InvariantCulture, $"{period.TotalMinutes:0.###} m"); + } + + return string.Create(CultureInfo.InvariantCulture, $"{period.TotalSeconds:0.###} s"); + } + + private static string RateLimitSourceText(ConcurrencyGroupInspectionSummary group) + => group.HasRateLimitOverride + ? group.OverrideRateLimit is null ? "Override: unlimited" : "Override" + : group.DefaultRateLimit is null ? "Built-in unlimited" : "Code default"; + + private static string RateLimitSourceTitle(ConcurrencyGroupInspectionSummary group) + => group.HasRateLimitOverride + ? group.OverrideRateLimit is null + ? "Live override disables the code-defined rate limit." + : string.Create(CultureInfo.InvariantCulture, $"Live rate override {FormatRateLimit(group.OverrideRateLimit)}.") + : group.DefaultRateLimit is null + ? "No start-rate limit is configured." + : string.Create(CultureInfo.InvariantCulture, $"Code default rate {FormatRateLimit(group.DefaultRateLimit)}."); + + private static string RateLimitSourceClass(ConcurrencyGroupInspectionSummary group) + => group.HasRateLimitOverride + ? "groups-source groups-source--override" + : group.DefaultRateLimit is not null + ? "groups-source groups-source--default" + : "groups-source"; + + private bool TryCreateRateLimit( + [NotNullWhen(true)] out ConcurrencyGroupRateLimit? rateLimit, + [NotNullWhen(false)] out string? validationMessage) + { + if (this._ratePermitInput <= 0) + { + rateLimit = null; + validationMessage = "Permit count must be positive."; + return false; + } + + if (this._ratePeriodInput <= 0) + { + rateLimit = null; + validationMessage = "Period must be positive."; + return false; + } + + try + { + var ticksPerUnit = this._ratePeriodUnit switch + { + HoursUnit => TimeSpan.TicksPerHour, + MinutesUnit => TimeSpan.TicksPerMinute, + _ => TimeSpan.TicksPerSecond, + }; + var periodTicks = decimal.ToInt64(this._ratePeriodInput * ticksPerUnit); + rateLimit = new ConcurrencyGroupRateLimit(this._ratePermitInput, TimeSpan.FromTicks(periodTicks)); + validationMessage = null; + return true; + } + catch (OverflowException) + { + rateLimit = null; + validationMessage = "Period is outside the supported range."; + return false; + } + } + + private static (decimal Value, string Unit) ToRatePeriodInput(TimeSpan period) + { + if (period.Ticks % TimeSpan.TicksPerHour == 0) + { + return ((decimal)period.Ticks / TimeSpan.TicksPerHour, HoursUnit); + } + + if (period.Ticks % TimeSpan.TicksPerMinute == 0) + { + return ((decimal)period.Ticks / TimeSpan.TicksPerMinute, MinutesUnit); + } + + return ((decimal)period.Ticks / TimeSpan.TicksPerSecond, SecondsUnit); + } + + private static string RatePermitInputLabel(ConcurrencyGroupInspectionSummary group) + => string.Create(CultureInfo.InvariantCulture, $"Rate permit count for concurrency group {group.GroupKey}"); + + private static string RatePeriodInputLabel(ConcurrencyGroupInspectionSummary group) + => string.Create(CultureInfo.InvariantCulture, $"Rate period for concurrency group {group.GroupKey}"); + + private static string RatePeriodUnitInputLabel(ConcurrencyGroupInspectionSummary group) + => string.Create(CultureInfo.InvariantCulture, $"Rate period unit for concurrency group {group.GroupKey}"); + + private static string EditRateLimitLabel(ConcurrencyGroupInspectionSummary group) + => string.Create(CultureInfo.InvariantCulture, $"Edit rate limit for concurrency group {group.GroupKey}"); + + private static string SaveRateLimitLabel(ConcurrencyGroupInspectionSummary group) + => string.Create(CultureInfo.InvariantCulture, $"Save rate limit for concurrency group {group.GroupKey}"); + + private static string CancelRateEditLabel(ConcurrencyGroupInspectionSummary group) + => string.Create(CultureInfo.InvariantCulture, $"Cancel rate-limit edit for concurrency group {group.GroupKey}"); + + private static string SetUnlimitedRateLimitLabel(ConcurrencyGroupInspectionSummary group) + => string.Create(CultureInfo.InvariantCulture, $"Set unlimited rate override for concurrency group {group.GroupKey}"); + + private static string ResetRateLimitLabel(ConcurrencyGroupInspectionSummary group) + => string.Create(CultureInfo.InvariantCulture, $"Reset rate-limit override for concurrency group {group.GroupKey}"); } diff --git a/src/Sheddueller.Dashboard/Components/_Imports.razor b/src/Sheddueller.Dashboard/Components/_Imports.razor index ab5fe9e..c3842c1 100644 --- a/src/Sheddueller.Dashboard/Components/_Imports.razor +++ b/src/Sheddueller.Dashboard/Components/_Imports.razor @@ -1,4 +1,5 @@ @using System.Globalization +@using System.Diagnostics.CodeAnalysis @using Microsoft.AspNetCore.Components @using Microsoft.AspNetCore.Components.Routing @using Microsoft.AspNetCore.Components.Web diff --git a/src/Sheddueller.Dashboard/Internal/DashboardFilters.cs b/src/Sheddueller.Dashboard/Internal/DashboardFilters.cs index 22a5b10..d1257f1 100644 --- a/src/Sheddueller.Dashboard/Internal/DashboardFilters.cs +++ b/src/Sheddueller.Dashboard/Internal/DashboardFilters.cs @@ -397,12 +397,17 @@ internal sealed class DashboardConcurrencyGroupFilters public bool HasBlockedJobsOnly { get; set; } + public bool RateLimitedOnly { get; set; } + public bool? SaturatedFilter => this.SaturatedOnly ? true : null; public bool? HasBlockedJobsFilter => this.HasBlockedJobsOnly ? true : null; + public bool? RateLimitedFilter + => this.RateLimitedOnly ? true : null; + public IReadOnlyList ApplyClientFilter( IReadOnlyList groups) => string.IsNullOrWhiteSpace(this.GroupKey) @@ -417,5 +422,6 @@ public ConcurrencyGroupInspectionQuery ToQuery( IsSaturated: this.SaturatedFilter, HasBlockedJobs: this.HasBlockedJobsFilter, PageSize: pageSize, - ContinuationToken: continuationToken); + ContinuationToken: continuationToken, + IsRateLimited: this.RateLimitedFilter); } diff --git a/src/Sheddueller.Postgres/Internal/Operations/ClearConcurrencyDefaultRateLimitOperation.cs b/src/Sheddueller.Postgres/Internal/Operations/ClearConcurrencyDefaultRateLimitOperation.cs new file mode 100644 index 0000000..9d0f249 --- /dev/null +++ b/src/Sheddueller.Postgres/Internal/Operations/ClearConcurrencyDefaultRateLimitOperation.cs @@ -0,0 +1,16 @@ +namespace Sheddueller.Postgres.Internal.Operations; + +using Sheddueller.Storage; + +internal static class ClearConcurrencyDefaultRateLimitOperation +{ + public static ValueTask ExecuteAsync( + PostgresOperationContext context, + ClearConcurrencyDefaultRateLimitRequest request, + CancellationToken cancellationToken) + => PostgresConcurrencyRateLimits.UpdateAsync( + context, + request.GroupKey, + current => current with { DefaultRateLimit = null }, + cancellationToken); +} diff --git a/src/Sheddueller.Postgres/Internal/Operations/ClearConcurrencyRateLimitOverrideOperation.cs b/src/Sheddueller.Postgres/Internal/Operations/ClearConcurrencyRateLimitOverrideOperation.cs new file mode 100644 index 0000000..9d743d1 --- /dev/null +++ b/src/Sheddueller.Postgres/Internal/Operations/ClearConcurrencyRateLimitOverrideOperation.cs @@ -0,0 +1,20 @@ +namespace Sheddueller.Postgres.Internal.Operations; + +using Sheddueller.Storage; + +internal static class ClearConcurrencyRateLimitOverrideOperation +{ + public static ValueTask ExecuteAsync( + PostgresOperationContext context, + ClearConcurrencyRateLimitOverrideRequest request, + CancellationToken cancellationToken) + => PostgresConcurrencyRateLimits.UpdateAsync( + context, + request.GroupKey, + current => current with + { + OverrideEnabled = false, + ConfiguredRateLimit = null, + }, + cancellationToken); +} diff --git a/src/Sheddueller.Postgres/Internal/Operations/PostgresConcurrencyGroupInspectionOperation.cs b/src/Sheddueller.Postgres/Internal/Operations/PostgresConcurrencyGroupInspectionOperation.cs index 9caf42c..cbd0f2e 100644 --- a/src/Sheddueller.Postgres/Internal/Operations/PostgresConcurrencyGroupInspectionOperation.cs +++ b/src/Sheddueller.Postgres/Internal/Operations/PostgresConcurrencyGroupInspectionOperation.cs @@ -40,7 +40,21 @@ public static async ValueTask SearchAsync( return new ConcurrencyGroupInspectionDetail( summary, await ReadClaimedJobIdsAsync(context, connection, groupKey, cancellationToken).ConfigureAwait(false), - await ReadBlockedJobIdsAsync(context, connection, groupKey, cancellationToken).ConfigureAwait(false)); + await ReadBlockedJobIdsAsync(context, connection, groupKey, BlockKind.Any, cancellationToken).ConfigureAwait(false)) + { + ConcurrencyBlockedJobIds = await ReadBlockedJobIdsAsync( + context, + connection, + groupKey, + BlockKind.Concurrency, + cancellationToken).ConfigureAwait(false), + RateBlockedJobIds = await ReadBlockedJobIdsAsync( + context, + connection, + groupKey, + BlockKind.Rate, + cancellationToken).ConfigureAwait(false), + }; } private static async ValueTask ReadTotalCountAsync( @@ -90,7 +104,18 @@ private static async ValueTask> summary.current_occupancy, summary.blocked_count, summary.is_saturated, - summary.updated_at_utc + summary.updated_at_utc, + summary.default_rate_permit_count, + summary.default_rate_period, + summary.rate_limit_override_enabled, + summary.configured_rate_permit_count, + summary.configured_rate_period, + summary.effective_rate_permit_count, + summary.effective_rate_period, + summary.rate_theoretical_arrival_at_utc, + summary.is_rate_limited, + summary.concurrency_blocked_count, + summary.rate_blocked_count from summary {CreateWhereClause(conditions)} order by summary.group_key asc @@ -118,7 +143,18 @@ order by summary.group_key asc summary.current_occupancy, summary.blocked_count, summary.is_saturated, - summary.updated_at_utc + summary.updated_at_utc, + summary.default_rate_permit_count, + summary.default_rate_period, + summary.rate_limit_override_enabled, + summary.configured_rate_permit_count, + summary.configured_rate_period, + summary.effective_rate_permit_count, + summary.effective_rate_period, + summary.rate_theoretical_arrival_at_utc, + summary.is_rate_limited, + summary.concurrency_blocked_count, + summary.rate_blocked_count from summary where summary.group_key = @group_key; """; @@ -150,6 +186,14 @@ private static async ValueTask> { DefaultLimit = defaultLimit, OverrideLimit = overrideLimit, + DefaultRateLimit = ReadRateLimit(reader, 8, 9), + HasRateLimitOverride = reader.GetBoolean(10), + OverrideRateLimit = ReadRateLimit(reader, 11, 12), + EffectiveRateLimit = ReadRateLimit(reader, 13, 14), + NextRatePermitAtUtc = reader.IsDBNull(15) ? null : PostgresConversion.ToDateTimeOffset(reader.GetValue(15)), + IsRateLimited = reader.GetBoolean(16), + ConcurrencyBlockedJobCount = Convert.ToInt32(reader.GetInt64(17), CultureInfo.InvariantCulture), + RateBlockedJobCount = Convert.ToInt32(reader.GetInt64(18), CultureInfo.InvariantCulture), }); } @@ -178,6 +222,12 @@ private static void ConfigureFilters( conditions.Add("(summary.blocked_count > 0) = @has_blocked_jobs"); command.Parameters.AddWithValue("has_blocked_jobs", hasBlockedJobs); } + + if (query.IsRateLimited is { } isRateLimited) + { + conditions.Add("summary.is_rate_limited = @is_rate_limited"); + command.Parameters.AddWithValue("is_rate_limited", isRateLimited); + } } private static string CreateWhereClause(List conditions) @@ -190,29 +240,47 @@ select group_key from {context.Names.ConcurrencyGroups} union select group_key from {context.Names.JobConcurrencyGroups} ), - blocked as ( - select job_group.group_key, count(*) as blocked_count - from {context.Names.JobConcurrencyGroups} job_group - join {context.Names.Jobs} job on job.job_id = job_group.job_id - left join {context.Names.ConcurrencyGroups} concurrency_group on concurrency_group.group_key = job_group.group_key - where job.state = 'Queued' - and (job.not_before_utc is null or job.not_before_utc <= transaction_timestamp()) - and coalesce(concurrency_group.in_use_count, 0) >= coalesce(concurrency_group.effective_limit, 1) - group by job_group.group_key - ), - summary as ( + group_state as ( select group_keys.group_key, concurrency_group.default_limit, concurrency_group.configured_limit as override_limit, coalesce(concurrency_group.effective_limit, 1) as effective_limit, coalesce(concurrency_group.in_use_count, 0) as current_occupancy, - coalesce(blocked.blocked_count, 0) as blocked_count, coalesce(concurrency_group.in_use_count, 0) >= coalesce(concurrency_group.effective_limit, 1) as is_saturated, + concurrency_group.default_rate_permit_count, + concurrency_group.default_rate_period, + coalesce(concurrency_group.rate_limit_override_enabled, false) as rate_limit_override_enabled, + concurrency_group.configured_rate_permit_count, + concurrency_group.configured_rate_period, + concurrency_group.effective_rate_permit_count, + concurrency_group.effective_rate_period, + concurrency_group.rate_theoretical_arrival_at_utc, + concurrency_group.effective_rate_permit_count is not null + and concurrency_group.rate_theoretical_arrival_at_utc > clock_timestamp() as is_rate_limited, concurrency_group.updated_at_utc from group_keys left join {context.Names.ConcurrencyGroups} concurrency_group on concurrency_group.group_key = group_keys.group_key - left join blocked on blocked.group_key = group_keys.group_key + ), + queued as ( + select job_group.group_key, count(*) as queued_count + from {context.Names.JobConcurrencyGroups} job_group + join {context.Names.Jobs} job on job.job_id = job_group.job_id + where job.state = 'Queued' + and (job.not_before_utc is null or job.not_before_utc <= transaction_timestamp()) + group by job_group.group_key + ), + summary as ( + select + group_state.*, + case + when group_state.is_saturated or group_state.is_rate_limited then coalesce(queued.queued_count, 0) + else 0 + end as blocked_count, + case when group_state.is_saturated then coalesce(queued.queued_count, 0) else 0 end as concurrency_blocked_count, + case when group_state.is_rate_limited then coalesce(queued.queued_count, 0) else 0 end as rate_blocked_count + from group_state + left join queued on queued.group_key = group_state.group_key ) """; @@ -239,6 +307,7 @@ private static async ValueTask> ReadBlockedJobIdsAsync( PostgresOperationContext context, NpgsqlConnection connection, string groupKey, + BlockKind blockKind, CancellationToken cancellationToken) => await ReadJobIdsAsync( connection, @@ -250,13 +319,38 @@ select job.job_id where job_group.group_key = @group_key and job.state = 'Queued' and (job.not_before_utc is null or job.not_before_utc <= transaction_timestamp()) - and coalesce(concurrency_group.in_use_count, 0) >= coalesce(concurrency_group.effective_limit, 1) + and ({CreateBlockCondition(blockKind)}) order by job.priority desc, job.enqueue_sequence asc; """, groupKey, cancellationToken) .ConfigureAwait(false); + private static string CreateBlockCondition(BlockKind blockKind) + => blockKind switch + { + BlockKind.Concurrency => "coalesce(concurrency_group.in_use_count, 0) >= coalesce(concurrency_group.effective_limit, 1)", + BlockKind.Rate => """ + concurrency_group.effective_rate_permit_count is not null + and concurrency_group.rate_theoretical_arrival_at_utc > clock_timestamp() + """, + _ => """ + coalesce(concurrency_group.in_use_count, 0) >= coalesce(concurrency_group.effective_limit, 1) + or ( + concurrency_group.effective_rate_permit_count is not null + and concurrency_group.rate_theoretical_arrival_at_utc > clock_timestamp() + ) + """, + }; + + private static ConcurrencyGroupRateLimit? ReadRateLimit( + NpgsqlDataReader reader, + int countOrdinal, + int periodOrdinal) + => reader.IsDBNull(countOrdinal) + ? null + : new ConcurrencyGroupRateLimit(reader.GetInt32(countOrdinal), reader.GetTimeSpan(periodOrdinal)); + private static async ValueTask> ReadJobIdsAsync( NpgsqlConnection connection, string commandText, @@ -289,4 +383,11 @@ private static void ValidateQuery(ConcurrencyGroupInspectionQuery query) throw new ArgumentException("Concurrency group inspection continuation token is invalid.", nameof(query)); } } + + private enum BlockKind + { + Any, + Concurrency, + Rate, + } } diff --git a/src/Sheddueller.Postgres/Internal/Operations/PostgresConcurrencyRateLimits.cs b/src/Sheddueller.Postgres/Internal/Operations/PostgresConcurrencyRateLimits.cs new file mode 100644 index 0000000..341103d --- /dev/null +++ b/src/Sheddueller.Postgres/Internal/Operations/PostgresConcurrencyRateLimits.cs @@ -0,0 +1,146 @@ +namespace Sheddueller.Postgres.Internal.Operations; + +using Npgsql; + +using NpgsqlTypes; + +internal static class PostgresConcurrencyRateLimits +{ + public static async ValueTask UpdateAsync( + PostgresOperationContext context, + string groupKey, + Func update, + CancellationToken cancellationToken) + { + await using var connection = await context.OpenConnectionAsync(cancellationToken).ConfigureAwait(false); + await using var transaction = await connection.BeginTransactionAsync(cancellationToken).ConfigureAwait(false); + + await PostgresOperationContext.ExecuteCountAsync( + connection, + transaction, + $""" + insert into {context.Names.ConcurrencyGroups} (group_key, configured_limit, in_use_count, updated_at_utc) + values (@group_key, null, 0, transaction_timestamp()) + on conflict (group_key) do nothing; + """, + command => command.Parameters.AddWithValue("group_key", groupKey), + cancellationToken) + .ConfigureAwait(false); + + var current = await ReadForUpdateAsync(context, connection, transaction, groupKey, cancellationToken).ConfigureAwait(false); + var next = update(current); + var resetRateState = current.EffectiveRateLimit != next.EffectiveRateLimit; + + await using var command = connection.CreateCommand(); + command.Transaction = transaction; + command.CommandText = + $""" + update {context.Names.ConcurrencyGroups} + set rate_limit_override_enabled = @override_enabled, + configured_rate_permit_count = @configured_permit_count, + configured_rate_period = @configured_period, + default_rate_permit_count = @default_permit_count, + default_rate_period = @default_period, + rate_theoretical_arrival_at_utc = case + when @reset_rate_state then null + else rate_theoretical_arrival_at_utc + end, + updated_at_utc = transaction_timestamp() + where group_key = @group_key; + """; + command.Parameters.AddWithValue("group_key", groupKey); + command.Parameters.AddWithValue("override_enabled", next.OverrideEnabled); + AddNullableInteger(command, "configured_permit_count", next.ConfiguredRateLimit?.PermitCount); + AddNullableInterval(command, "configured_period", next.ConfiguredRateLimit?.Period); + AddNullableInteger(command, "default_permit_count", next.DefaultRateLimit?.PermitCount); + AddNullableInterval(command, "default_period", next.DefaultRateLimit?.Period); + command.Parameters.AddWithValue("reset_rate_state", resetRateState); + await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + + await context.NotifyAsync(connection, transaction, cancellationToken).ConfigureAwait(false); + await transaction.CommitAsync(cancellationToken).ConfigureAwait(false); + } + + public static async ValueTask GetOverrideAsync( + PostgresOperationContext context, + string groupKey, + CancellationToken cancellationToken) + { + await using var connection = await context.OpenConnectionAsync(cancellationToken).ConfigureAwait(false); + await using var command = connection.CreateCommand(); + command.CommandText = + $""" + select rate_limit_override_enabled, configured_rate_permit_count, configured_rate_period + from {context.Names.ConcurrencyGroups} + where group_key = @group_key; + """; + command.Parameters.AddWithValue("group_key", groupKey); + + await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false) || !reader.GetBoolean(0)) + { + return new ConcurrencyGroupRateLimitOverride(ConcurrencyGroupRateLimitOverrideKind.Inherit); + } + + return reader.IsDBNull(1) + ? new ConcurrencyGroupRateLimitOverride(ConcurrencyGroupRateLimitOverrideKind.Unlimited) + : new ConcurrencyGroupRateLimitOverride( + ConcurrencyGroupRateLimitOverrideKind.Limited, + new ConcurrencyGroupRateLimit(reader.GetInt32(1), reader.GetTimeSpan(2))); + } + + private static async ValueTask ReadForUpdateAsync( + PostgresOperationContext context, + NpgsqlConnection connection, + NpgsqlTransaction transaction, + string groupKey, + CancellationToken cancellationToken) + { + await using var command = connection.CreateCommand(); + command.Transaction = transaction; + command.CommandText = + $""" + select + rate_limit_override_enabled, + configured_rate_permit_count, + configured_rate_period, + default_rate_permit_count, + default_rate_period + from {context.Names.ConcurrencyGroups} + where group_key = @group_key + for update; + """; + command.Parameters.AddWithValue("group_key", groupKey); + + await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + { + throw new InvalidOperationException($"Concurrency group '{groupKey}' could not be initialized."); + } + + return new Configuration( + reader.GetBoolean(0), + ReadRateLimit(reader, 1, 2), + ReadRateLimit(reader, 3, 4)); + } + + private static ConcurrencyGroupRateLimit? ReadRateLimit(NpgsqlDataReader reader, int countOrdinal, int periodOrdinal) + => reader.IsDBNull(countOrdinal) + ? null + : new ConcurrencyGroupRateLimit(reader.GetInt32(countOrdinal), reader.GetTimeSpan(periodOrdinal)); + + private static void AddNullableInteger(NpgsqlCommand command, string name, int? value) + => command.Parameters.AddWithValue(name, NpgsqlDbType.Integer, value is null ? DBNull.Value : value.Value); + + private static void AddNullableInterval(NpgsqlCommand command, string name, TimeSpan? value) + => command.Parameters.AddWithValue(name, NpgsqlDbType.Interval, value is null ? DBNull.Value : value.Value); + + internal sealed record Configuration( + bool OverrideEnabled, + ConcurrencyGroupRateLimit? ConfiguredRateLimit, + ConcurrencyGroupRateLimit? DefaultRateLimit) + { + public ConcurrencyGroupRateLimit? EffectiveRateLimit + => this.OverrideEnabled ? this.ConfiguredRateLimit : this.DefaultRateLimit; + } +} diff --git a/src/Sheddueller.Postgres/Internal/Operations/PostgresJobGroups.cs b/src/Sheddueller.Postgres/Internal/Operations/PostgresJobGroups.cs index 3460c06..a1f52c6 100644 --- a/src/Sheddueller.Postgres/Internal/Operations/PostgresJobGroups.cs +++ b/src/Sheddueller.Postgres/Internal/Operations/PostgresJobGroups.cs @@ -71,7 +71,13 @@ public static async ValueTask TryReserveGroupsAsync( command.Transaction = transaction; command.CommandText = $""" - select group_key, effective_limit, in_use_count + select + group_key, + effective_limit, + in_use_count, + effective_rate_permit_count, + rate_theoretical_arrival_at_utc is null + or rate_theoretical_arrival_at_utc <= clock_timestamp() as rate_available from {context.Names.ConcurrencyGroups} where group_key = any(@group_keys) order by group_key asc @@ -87,7 +93,8 @@ order by group_key asc lockCount++; var effectiveLimit = reader.GetInt32(1); var inUseCount = reader.GetInt32(2); - if (inUseCount >= effectiveLimit) + var rateAvailable = reader.IsDBNull(3) || reader.GetBoolean(4); + if (inUseCount >= effectiveLimit || !rateAvailable) { return false; } @@ -105,6 +112,13 @@ await PostgresOperationContext.ExecuteCountAsync( $""" update {context.Names.ConcurrencyGroups} set in_use_count = in_use_count + 1, + rate_theoretical_arrival_at_utc = case + when effective_rate_permit_count is null then null + else greatest( + coalesce(rate_theoretical_arrival_at_utc, clock_timestamp()), + clock_timestamp()) + + (effective_rate_period / effective_rate_permit_count) + end, updated_at_utc = transaction_timestamp() where group_key = any(@group_keys); """, diff --git a/src/Sheddueller.Postgres/Internal/Operations/SetConcurrencyDefaultRateLimitOperation.cs b/src/Sheddueller.Postgres/Internal/Operations/SetConcurrencyDefaultRateLimitOperation.cs new file mode 100644 index 0000000..2c7bc19 --- /dev/null +++ b/src/Sheddueller.Postgres/Internal/Operations/SetConcurrencyDefaultRateLimitOperation.cs @@ -0,0 +1,16 @@ +namespace Sheddueller.Postgres.Internal.Operations; + +using Sheddueller.Storage; + +internal static class SetConcurrencyDefaultRateLimitOperation +{ + public static ValueTask ExecuteAsync( + PostgresOperationContext context, + SetConcurrencyDefaultRateLimitRequest request, + CancellationToken cancellationToken) + => PostgresConcurrencyRateLimits.UpdateAsync( + context, + request.GroupKey, + current => current with { DefaultRateLimit = request.RateLimit }, + cancellationToken); +} diff --git a/src/Sheddueller.Postgres/Internal/Operations/SetConcurrencyRateLimitOperation.cs b/src/Sheddueller.Postgres/Internal/Operations/SetConcurrencyRateLimitOperation.cs new file mode 100644 index 0000000..97fb002 --- /dev/null +++ b/src/Sheddueller.Postgres/Internal/Operations/SetConcurrencyRateLimitOperation.cs @@ -0,0 +1,20 @@ +namespace Sheddueller.Postgres.Internal.Operations; + +using Sheddueller.Storage; + +internal static class SetConcurrencyRateLimitOperation +{ + public static ValueTask ExecuteAsync( + PostgresOperationContext context, + SetConcurrencyRateLimitRequest request, + CancellationToken cancellationToken) + => PostgresConcurrencyRateLimits.UpdateAsync( + context, + request.GroupKey, + current => current with + { + OverrideEnabled = true, + ConfiguredRateLimit = request.RateLimit, + }, + cancellationToken); +} diff --git a/src/Sheddueller.Postgres/Internal/Operations/SetConcurrencyUnlimitedRateLimitOperation.cs b/src/Sheddueller.Postgres/Internal/Operations/SetConcurrencyUnlimitedRateLimitOperation.cs new file mode 100644 index 0000000..0e4a95c --- /dev/null +++ b/src/Sheddueller.Postgres/Internal/Operations/SetConcurrencyUnlimitedRateLimitOperation.cs @@ -0,0 +1,20 @@ +namespace Sheddueller.Postgres.Internal.Operations; + +using Sheddueller.Storage; + +internal static class SetConcurrencyUnlimitedRateLimitOperation +{ + public static ValueTask ExecuteAsync( + PostgresOperationContext context, + SetConcurrencyUnlimitedRateLimitRequest request, + CancellationToken cancellationToken) + => PostgresConcurrencyRateLimits.UpdateAsync( + context, + request.GroupKey, + current => current with + { + OverrideEnabled = true, + ConfiguredRateLimit = null, + }, + cancellationToken); +} diff --git a/src/Sheddueller.Postgres/Internal/Operations/TryClaimNextJobOperation.cs b/src/Sheddueller.Postgres/Internal/Operations/TryClaimNextJobOperation.cs index 8a9ef2d..1f7e509 100644 --- a/src/Sheddueller.Postgres/Internal/Operations/TryClaimNextJobOperation.cs +++ b/src/Sheddueller.Postgres/Internal/Operations/TryClaimNextJobOperation.cs @@ -97,8 +97,9 @@ await PostgresJobEvents.AppendAndNotifyInTransactionAsync( return new ClaimJobResult.Claimed(claimed); } + var nextClaimAtUtc = await ReadNextRateLimitedClaimAtAsync(context, connection, transaction, cancellationToken).ConfigureAwait(false); await transaction.CommitAsync(cancellationToken).ConfigureAwait(false); - return new ClaimJobResult.NoJobAvailable(); + return new ClaimJobResult.NoJobAvailable(nextClaimAtUtc); } private static async ValueTask> ReadClaimCandidatesAsync( @@ -120,7 +121,13 @@ select 1 from {context.Names.JobConcurrencyGroups} job_group join {context.Names.ConcurrencyGroups} concurrency_group on concurrency_group.group_key = job_group.group_key where job_group.job_id = job.job_id - and concurrency_group.in_use_count >= concurrency_group.effective_limit + and ( + concurrency_group.in_use_count >= concurrency_group.effective_limit + or ( + concurrency_group.effective_rate_permit_count is not null + and concurrency_group.rate_theoretical_arrival_at_utc > clock_timestamp() + ) + ) ) order by job.priority desc, job.enqueue_sequence asc for update of job skip locked @@ -137,4 +144,32 @@ select 1 return jobIds; } + + private static async ValueTask ReadNextRateLimitedClaimAtAsync( + PostgresOperationContext context, + NpgsqlConnection connection, + NpgsqlTransaction transaction, + CancellationToken cancellationToken) + { + await using var command = connection.CreateCommand(); + command.Transaction = transaction; + command.CommandText = + $""" + select min(rate_blocked.next_claim_at_utc) + from ( + select job.job_id, max(concurrency_group.rate_theoretical_arrival_at_utc) as next_claim_at_utc + from {context.Names.Jobs} job + join {context.Names.JobConcurrencyGroups} job_group on job_group.job_id = job.job_id + join {context.Names.ConcurrencyGroups} concurrency_group on concurrency_group.group_key = job_group.group_key + where job.state = 'Queued' + and (job.not_before_utc is null or job.not_before_utc <= transaction_timestamp()) + and concurrency_group.effective_rate_permit_count is not null + and concurrency_group.rate_theoretical_arrival_at_utc > clock_timestamp() + group by job.job_id + ) rate_blocked; + """; + + var value = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false); + return value is null or DBNull ? null : PostgresConversion.ToDateTimeOffset(value); + } } diff --git a/src/Sheddueller.Postgres/Internal/PostgresJobStore.cs b/src/Sheddueller.Postgres/Internal/PostgresJobStore.cs index bf26f30..d8aa16e 100644 --- a/src/Sheddueller.Postgres/Internal/PostgresJobStore.cs +++ b/src/Sheddueller.Postgres/Internal/PostgresJobStore.cs @@ -177,6 +177,56 @@ public ValueTask ClearConcurrencyLimitOverrideAsync( CancellationToken cancellationToken = default) => GetConfiguredConcurrencyLimitOperation.ExecuteAsync(this._context, groupKey, cancellationToken); + public ValueTask SetConcurrencyRateLimitAsync( + SetConcurrencyRateLimitRequest request, + CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(request); + + return SetConcurrencyRateLimitOperation.ExecuteAsync(this._context, request, cancellationToken); + } + + public ValueTask SetConcurrencyDefaultRateLimitAsync( + SetConcurrencyDefaultRateLimitRequest request, + CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(request); + + return SetConcurrencyDefaultRateLimitOperation.ExecuteAsync(this._context, request, cancellationToken); + } + + public ValueTask ClearConcurrencyDefaultRateLimitAsync( + ClearConcurrencyDefaultRateLimitRequest request, + CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(request); + + return ClearConcurrencyDefaultRateLimitOperation.ExecuteAsync(this._context, request, cancellationToken); + } + + public ValueTask SetConcurrencyUnlimitedRateLimitAsync( + SetConcurrencyUnlimitedRateLimitRequest request, + CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(request); + + return SetConcurrencyUnlimitedRateLimitOperation.ExecuteAsync(this._context, request, cancellationToken); + } + + public ValueTask ClearConcurrencyRateLimitOverrideAsync( + ClearConcurrencyRateLimitOverrideRequest request, + CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(request); + + return ClearConcurrencyRateLimitOverrideOperation.ExecuteAsync(this._context, request, cancellationToken); + } + + public ValueTask GetConcurrencyRateLimitOverrideAsync( + string groupKey, + CancellationToken cancellationToken = default) + => PostgresConcurrencyRateLimits.GetOverrideAsync(this._context, groupKey, cancellationToken); + public ValueTask CreateOrUpdateRecurringScheduleAsync( UpsertRecurringScheduleRequest request, CancellationToken cancellationToken = default) diff --git a/src/Sheddueller.Postgres/Internal/PostgresMigrator.cs b/src/Sheddueller.Postgres/Internal/PostgresMigrator.cs index 1979dc8..df019d5 100644 --- a/src/Sheddueller.Postgres/Internal/PostgresMigrator.cs +++ b/src/Sheddueller.Postgres/Internal/PostgresMigrator.cs @@ -176,10 +176,36 @@ alter table {this._names.JobTags} configured_limit integer null, default_limit integer null, effective_limit integer generated always as (coalesce(configured_limit, default_limit, 1)) stored, + rate_limit_override_enabled boolean not null default false, + configured_rate_permit_count integer null, + configured_rate_period interval null, + default_rate_permit_count integer null, + default_rate_period interval null, + effective_rate_permit_count integer generated always as ( + case when rate_limit_override_enabled then configured_rate_permit_count else default_rate_permit_count end + ) stored, + effective_rate_period interval generated always as ( + case when rate_limit_override_enabled then configured_rate_period else default_rate_period end + ) stored, + rate_theoretical_arrival_at_utc timestamptz null, in_use_count integer not null, updated_at_utc timestamptz not null, constraint concurrency_groups_configured_limit_check check (configured_limit is null or configured_limit > 0), constraint concurrency_groups_default_limit_check check (default_limit is null or default_limit > 0), + constraint concurrency_groups_configured_rate_check check ( + (not rate_limit_override_enabled and configured_rate_permit_count is null and configured_rate_period is null) + or ( + rate_limit_override_enabled + and ( + (configured_rate_permit_count is null and configured_rate_period is null) + or (configured_rate_permit_count > 0 and configured_rate_period > interval '0') + ) + ) + ), + constraint concurrency_groups_default_rate_check check ( + (default_rate_permit_count is null and default_rate_period is null) + or (default_rate_permit_count > 0 and default_rate_period > interval '0') + ), constraint concurrency_groups_in_use_count_check check (in_use_count >= 0) ); @@ -189,11 +215,41 @@ alter table {this._names.ConcurrencyGroups} alter table {this._names.ConcurrencyGroups} add column if not exists effective_limit integer generated always as (coalesce(configured_limit, default_limit, 1)) stored; + alter table {this._names.ConcurrencyGroups} + add column if not exists rate_limit_override_enabled boolean not null default false, + add column if not exists configured_rate_permit_count integer null, + add column if not exists configured_rate_period interval null, + add column if not exists default_rate_permit_count integer null, + add column if not exists default_rate_period interval null, + add column if not exists effective_rate_permit_count integer generated always as ( + case when rate_limit_override_enabled then configured_rate_permit_count else default_rate_permit_count end + ) stored, + add column if not exists effective_rate_period interval generated always as ( + case when rate_limit_override_enabled then configured_rate_period else default_rate_period end + ) stored, + add column if not exists rate_theoretical_arrival_at_utc timestamptz null; + alter table {this._names.ConcurrencyGroups} drop constraint if exists concurrency_groups_configured_limit_check, add constraint concurrency_groups_configured_limit_check check (configured_limit is null or configured_limit > 0), drop constraint if exists concurrency_groups_default_limit_check, add constraint concurrency_groups_default_limit_check check (default_limit is null or default_limit > 0), + drop constraint if exists concurrency_groups_configured_rate_check, + add constraint concurrency_groups_configured_rate_check check ( + (not rate_limit_override_enabled and configured_rate_permit_count is null and configured_rate_period is null) + or ( + rate_limit_override_enabled + and ( + (configured_rate_permit_count is null and configured_rate_period is null) + or (configured_rate_permit_count > 0 and configured_rate_period > interval '0') + ) + ) + ), + drop constraint if exists concurrency_groups_default_rate_check, + add constraint concurrency_groups_default_rate_check check ( + (default_rate_permit_count is null and default_rate_period is null) + or (default_rate_permit_count > 0 and default_rate_period > interval '0') + ), drop constraint if exists concurrency_groups_in_use_count_check, add constraint concurrency_groups_in_use_count_check check (in_use_count >= 0); diff --git a/src/Sheddueller.Postgres/Internal/PostgresNames.cs b/src/Sheddueller.Postgres/Internal/PostgresNames.cs index 00089d8..be0a4ac 100644 --- a/src/Sheddueller.Postgres/Internal/PostgresNames.cs +++ b/src/Sheddueller.Postgres/Internal/PostgresNames.cs @@ -4,7 +4,7 @@ namespace Sheddueller.Postgres.Internal; internal sealed class PostgresNames { - public const int ExpectedSchemaVersion = 13; + public const int ExpectedSchemaVersion = 14; public const string WakeupChannel = "sheddueller_wakeup"; public const string JobEventChannel = "sheddueller_job_event"; diff --git a/src/Sheddueller.Worker/Internal/ShedduellerWorker.cs b/src/Sheddueller.Worker/Internal/ShedduellerWorker.cs index b38203c..6fad859 100644 --- a/src/Sheddueller.Worker/Internal/ShedduellerWorker.cs +++ b/src/Sheddueller.Worker/Internal/ShedduellerWorker.cs @@ -50,6 +50,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) await this.RunPeriodicStoreWorkAsync(store, stoppingToken).ConfigureAwait(false); var claimedJob = false; + DateTimeOffset? nextClaimAtUtc = null; while (!stoppingToken.IsCancellationRequested && this._runningJobs.Count < this._options.Value.MaxConcurrentExecutionsPerNode) { var now = this._timeProvider.GetUtcNow(); @@ -59,6 +60,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) if (claimResult is not ClaimJobResult.Claimed claimed) { + nextClaimAtUtc = ((ClaimJobResult.NoJobAvailable)claimResult).NextClaimAtUtc; break; } @@ -72,7 +74,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) continue; } - await this.WaitForWorkOrCapacityAsync(stoppingToken).ConfigureAwait(false); + await this.WaitForWorkOrCapacityAsync(nextClaimAtUtc, stoppingToken).ConfigureAwait(false); } } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) @@ -89,16 +91,27 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) this._logger.WorkerStopped(this._nodeIdProvider.NodeId); } - private async ValueTask WaitForWorkOrCapacityAsync(CancellationToken stoppingToken) + private async ValueTask WaitForWorkOrCapacityAsync( + DateTimeOffset? nextClaimAtUtc, + CancellationToken stoppingToken) { + var timeout = CalculateWaitTimeout( + this._options.Value.IdlePollingInterval, + this._timeProvider.GetUtcNow(), + nextClaimAtUtc); + if (timeout <= TimeSpan.Zero) + { + return; + } + if (this._runningJobs.IsEmpty) { - await this._wakeSignal.WaitAsync(this._options.Value.IdlePollingInterval, stoppingToken).ConfigureAwait(false); + await this._wakeSignal.WaitAsync(timeout, stoppingToken).ConfigureAwait(false); return; } - var delayTask = Task.Delay(this._options.Value.IdlePollingInterval, stoppingToken); - var signalTask = this._wakeSignal.WaitAsync(this._options.Value.IdlePollingInterval, stoppingToken).AsTask(); + var delayTask = Task.Delay(timeout, stoppingToken); + var signalTask = this._wakeSignal.WaitAsync(timeout, stoppingToken).AsTask(); var completedRunningTask = await Task.WhenAny(this._runningJobs.Keys.Append(delayTask).Append(signalTask)).ConfigureAwait(false); if (completedRunningTask == signalTask) @@ -107,6 +120,22 @@ private async ValueTask WaitForWorkOrCapacityAsync(CancellationToken stoppingTok } } + internal static TimeSpan CalculateWaitTimeout( + TimeSpan idlePollingInterval, + DateTimeOffset nowUtc, + DateTimeOffset? nextClaimAtUtc) + { + if (nextClaimAtUtc is not { } nextClaim) + { + return idlePollingInterval; + } + + var rateDelay = nextClaim - nowUtc; + return rateDelay <= TimeSpan.Zero + ? TimeSpan.Zero + : TimeSpan.FromTicks(Math.Min(idlePollingInterval.Ticks, rateDelay.Ticks)); + } + [SuppressMessage("Design", "CA1031:Do not catch general exception types", Justification = "Job failures must be persisted instead of escaping the worker.")] private async Task ExecuteClaimedJobAsync(IJobStore store, ClaimedJob job, CancellationToken stoppingToken) { diff --git a/src/Sheddueller/ConcurrencyGroupRateLimit.cs b/src/Sheddueller/ConcurrencyGroupRateLimit.cs new file mode 100644 index 0000000..13eeca1 --- /dev/null +++ b/src/Sheddueller/ConcurrencyGroupRateLimit.cs @@ -0,0 +1,10 @@ +namespace Sheddueller; + +/// +/// Defines a smooth concurrency-group job-start rate. +/// +/// The number of job starts permitted during . +/// The period over which starts are evenly spaced. +public sealed record ConcurrencyGroupRateLimit( + int PermitCount, + TimeSpan Period); diff --git a/src/Sheddueller/ConcurrencyGroupRateLimitOverride.cs b/src/Sheddueller/ConcurrencyGroupRateLimitOverride.cs new file mode 100644 index 0000000..7d91847 --- /dev/null +++ b/src/Sheddueller/ConcurrencyGroupRateLimitOverride.cs @@ -0,0 +1,10 @@ +namespace Sheddueller; + +/// +/// Describes the live rate-limit override configured for a concurrency group. +/// +/// The override kind. +/// The configured rate when is . +public sealed record ConcurrencyGroupRateLimitOverride( + ConcurrencyGroupRateLimitOverrideKind Kind, + ConcurrencyGroupRateLimit? RateLimit = null); diff --git a/src/Sheddueller/ConcurrencyGroupRateLimitOverrideKind.cs b/src/Sheddueller/ConcurrencyGroupRateLimitOverrideKind.cs new file mode 100644 index 0000000..b8717bc --- /dev/null +++ b/src/Sheddueller/ConcurrencyGroupRateLimitOverrideKind.cs @@ -0,0 +1,22 @@ +namespace Sheddueller; + +/// +/// Describes the live rate-limit override configured for a concurrency group. +/// +public enum ConcurrencyGroupRateLimitOverrideKind +{ + /// + /// The group inherits its code-defined default rate. + /// + Inherit, + + /// + /// The group has a limited live rate override. + /// + Limited, + + /// + /// The group has an explicitly unlimited live rate override. + /// + Unlimited, +} diff --git a/src/Sheddueller/IConcurrencyGroupManager.cs b/src/Sheddueller/IConcurrencyGroupManager.cs index aab183a..d64f93a 100644 --- a/src/Sheddueller/IConcurrencyGroupManager.cs +++ b/src/Sheddueller/IConcurrencyGroupManager.cs @@ -34,4 +34,54 @@ ValueTask ClearLimitOverrideAsync( ValueTask GetConfiguredLimitAsync( string groupKey, CancellationToken cancellationToken = default); + + /// + /// Sets the live rate-limit override for a concurrency group. + /// + ValueTask SetRateLimitAsync( + string groupKey, + ConcurrencyGroupRateLimit rateLimit, + CancellationToken cancellationToken = default) + => throw new NotSupportedException("This concurrency-group manager does not support rate limits."); + + /// + /// Sets the code-defined default rate limit for a concurrency group without clearing a live override. + /// + ValueTask SetDefaultRateLimitAsync( + string groupKey, + ConcurrencyGroupRateLimit rateLimit, + CancellationToken cancellationToken = default) + => throw new NotSupportedException("This concurrency-group manager does not support rate limits."); + + /// + /// Clears the code-defined default rate limit for a concurrency group without clearing a live override. + /// + ValueTask ClearDefaultRateLimitAsync( + string groupKey, + CancellationToken cancellationToken = default) + => throw new NotSupportedException("This concurrency-group manager does not support rate limits."); + + /// + /// Sets an explicitly unlimited live rate override for a concurrency group. + /// + ValueTask SetUnlimitedRateLimitAsync( + string groupKey, + CancellationToken cancellationToken = default) + => throw new NotSupportedException("This concurrency-group manager does not support rate limits."); + + /// + /// Clears the live rate-limit override for a concurrency group, falling back to its code-defined default. + /// + ValueTask ClearRateLimitOverrideAsync( + string groupKey, + CancellationToken cancellationToken = default) + => throw new NotSupportedException("This concurrency-group manager does not support rate limits."); + + /// + /// Gets the live rate-limit override for a concurrency group. + /// + ValueTask GetRateLimitOverrideAsync( + string groupKey, + CancellationToken cancellationToken = default) + => ValueTask.FromResult(new ConcurrencyGroupRateLimitOverride(ConcurrencyGroupRateLimitOverrideKind.Inherit)); } diff --git a/src/Sheddueller/Inspection/ConcurrencyGroups/ConcurrencyGroupInspectionDetail.cs b/src/Sheddueller/Inspection/ConcurrencyGroups/ConcurrencyGroupInspectionDetail.cs index 4a6a890..b078c05 100644 --- a/src/Sheddueller/Inspection/ConcurrencyGroups/ConcurrencyGroupInspectionDetail.cs +++ b/src/Sheddueller/Inspection/ConcurrencyGroups/ConcurrencyGroupInspectionDetail.cs @@ -6,4 +6,15 @@ namespace Sheddueller.Inspection.ConcurrencyGroups; public sealed record ConcurrencyGroupInspectionDetail( ConcurrencyGroupInspectionSummary Summary, IReadOnlyList ClaimedJobIds, - IReadOnlyList BlockedJobIds); + IReadOnlyList BlockedJobIds) +{ + /// + /// Gets due queued jobs blocked by the concurrency limit. + /// + public IReadOnlyList ConcurrencyBlockedJobIds { get; init; } = []; + + /// + /// Gets due queued jobs blocked by the start-rate limit. + /// + public IReadOnlyList RateBlockedJobIds { get; init; } = []; +} diff --git a/src/Sheddueller/Inspection/ConcurrencyGroups/ConcurrencyGroupInspectionQuery.cs b/src/Sheddueller/Inspection/ConcurrencyGroups/ConcurrencyGroupInspectionQuery.cs index b595832..bf84ecc 100644 --- a/src/Sheddueller/Inspection/ConcurrencyGroups/ConcurrencyGroupInspectionQuery.cs +++ b/src/Sheddueller/Inspection/ConcurrencyGroups/ConcurrencyGroupInspectionQuery.cs @@ -8,4 +8,5 @@ public sealed record ConcurrencyGroupInspectionQuery( bool? IsSaturated = null, bool? HasBlockedJobs = null, int PageSize = 100, - string? ContinuationToken = null); + string? ContinuationToken = null, + bool? IsRateLimited = null); diff --git a/src/Sheddueller/Inspection/ConcurrencyGroups/ConcurrencyGroupInspectionSummary.cs b/src/Sheddueller/Inspection/ConcurrencyGroups/ConcurrencyGroupInspectionSummary.cs index 71fe24e..9d8e8fd 100644 --- a/src/Sheddueller/Inspection/ConcurrencyGroups/ConcurrencyGroupInspectionSummary.cs +++ b/src/Sheddueller/Inspection/ConcurrencyGroups/ConcurrencyGroupInspectionSummary.cs @@ -20,4 +20,44 @@ public sealed record ConcurrencyGroupInspectionSummary( /// Gets the live override limit, if one exists. /// public int? OverrideLimit { get; init; } + + /// + /// Gets the code-defined default start rate, if one exists. + /// + public ConcurrencyGroupRateLimit? DefaultRateLimit { get; init; } + + /// + /// Gets whether a live start-rate override exists. + /// + public bool HasRateLimitOverride { get; init; } + + /// + /// Gets the live start-rate override. Null with set means explicitly unlimited. + /// + public ConcurrencyGroupRateLimit? OverrideRateLimit { get; init; } + + /// + /// Gets the effective start rate, or null when starts are unlimited. + /// + public ConcurrencyGroupRateLimit? EffectiveRateLimit { get; init; } + + /// + /// Gets the theoretical next permitted start time, if rate state has been consumed. + /// + public DateTimeOffset? NextRatePermitAtUtc { get; init; } + + /// + /// Gets whether the group is currently waiting for its next rate permit. + /// + public bool IsRateLimited { get; init; } + + /// + /// Gets the number of due queued jobs blocked by the concurrency limit. + /// + public int ConcurrencyBlockedJobCount { get; init; } + + /// + /// Gets the number of due queued jobs blocked by the start-rate limit. + /// + public int RateBlockedJobCount { get; init; } } diff --git a/src/Sheddueller/Logging/ShedduellerLoggerMessages.cs b/src/Sheddueller/Logging/ShedduellerLoggerMessages.cs index a873267..4ace03b 100644 --- a/src/Sheddueller/Logging/ShedduellerLoggerMessages.cs +++ b/src/Sheddueller/Logging/ShedduellerLoggerMessages.cs @@ -119,6 +119,50 @@ public static partial void ConcurrencyGroupLimitOverrideCleared( this ILogger logger, string groupKey); + [LoggerMessage( + EventIdStart + 33, + LogLevel.Debug, + "Set concurrency group {GroupKey} rate limit to {PermitCount} starts per {Period}.")] + public static partial void ConcurrencyGroupRateLimitSet( + this ILogger logger, + string groupKey, + int permitCount, + TimeSpan period); + + [LoggerMessage( + EventIdStart + 34, + LogLevel.Debug, + "Set concurrency group {GroupKey} default rate limit to {PermitCount} starts per {Period}.")] + public static partial void ConcurrencyGroupDefaultRateLimitSet( + this ILogger logger, + string groupKey, + int permitCount, + TimeSpan period); + + [LoggerMessage( + EventIdStart + 35, + LogLevel.Debug, + "Cleared concurrency group {GroupKey} default rate limit.")] + public static partial void ConcurrencyGroupDefaultRateLimitCleared( + this ILogger logger, + string groupKey); + + [LoggerMessage( + EventIdStart + 36, + LogLevel.Debug, + "Set concurrency group {GroupKey} live rate override to unlimited.")] + public static partial void ConcurrencyGroupUnlimitedRateLimitSet( + this ILogger logger, + string groupKey); + + [LoggerMessage( + EventIdStart + 37, + LogLevel.Debug, + "Cleared concurrency group {GroupKey} rate-limit override.")] + public static partial void ConcurrencyGroupRateLimitOverrideCleared( + this ILogger logger, + string groupKey); + [LoggerMessage( EventIdStart + 40, LogLevel.Warning, diff --git a/src/Sheddueller/Runtime/ConcurrencyGroupManager.cs b/src/Sheddueller/Runtime/ConcurrencyGroupManager.cs index a8da76e..7396400 100644 --- a/src/Sheddueller/Runtime/ConcurrencyGroupManager.cs +++ b/src/Sheddueller/Runtime/ConcurrencyGroupManager.cs @@ -53,6 +53,92 @@ await store return store.GetConfiguredConcurrencyLimitAsync(groupKey, cancellationToken); } + public async ValueTask SetRateLimitAsync( + string groupKey, + ConcurrencyGroupRateLimit rateLimit, + CancellationToken cancellationToken = default) + { + SubmissionValidator.ValidateConcurrencyGroupKey(groupKey); + ValidateRateLimit(rateLimit); + + await store + .SetConcurrencyRateLimitAsync(new SetConcurrencyRateLimitRequest(groupKey, rateLimit, timeProvider.GetUtcNow()), cancellationToken) + .ConfigureAwait(false); + wakeSignal.Notify(); + logger.ConcurrencyGroupRateLimitSet(groupKey, rateLimit.PermitCount, rateLimit.Period); + } + + public async ValueTask SetDefaultRateLimitAsync( + string groupKey, + ConcurrencyGroupRateLimit rateLimit, + CancellationToken cancellationToken = default) + { + SubmissionValidator.ValidateConcurrencyGroupKey(groupKey); + ValidateRateLimit(rateLimit); + + await store + .SetConcurrencyDefaultRateLimitAsync( + new SetConcurrencyDefaultRateLimitRequest(groupKey, rateLimit, timeProvider.GetUtcNow()), + cancellationToken) + .ConfigureAwait(false); + wakeSignal.Notify(); + logger.ConcurrencyGroupDefaultRateLimitSet(groupKey, rateLimit.PermitCount, rateLimit.Period); + } + + public async ValueTask ClearDefaultRateLimitAsync( + string groupKey, + CancellationToken cancellationToken = default) + { + SubmissionValidator.ValidateConcurrencyGroupKey(groupKey); + + await store + .ClearConcurrencyDefaultRateLimitAsync( + new ClearConcurrencyDefaultRateLimitRequest(groupKey, timeProvider.GetUtcNow()), + cancellationToken) + .ConfigureAwait(false); + wakeSignal.Notify(); + logger.ConcurrencyGroupDefaultRateLimitCleared(groupKey); + } + + public async ValueTask SetUnlimitedRateLimitAsync( + string groupKey, + CancellationToken cancellationToken = default) + { + SubmissionValidator.ValidateConcurrencyGroupKey(groupKey); + + await store + .SetConcurrencyUnlimitedRateLimitAsync( + new SetConcurrencyUnlimitedRateLimitRequest(groupKey, timeProvider.GetUtcNow()), + cancellationToken) + .ConfigureAwait(false); + wakeSignal.Notify(); + logger.ConcurrencyGroupUnlimitedRateLimitSet(groupKey); + } + + public async ValueTask ClearRateLimitOverrideAsync( + string groupKey, + CancellationToken cancellationToken = default) + { + SubmissionValidator.ValidateConcurrencyGroupKey(groupKey); + + await store + .ClearConcurrencyRateLimitOverrideAsync( + new ClearConcurrencyRateLimitOverrideRequest(groupKey, timeProvider.GetUtcNow()), + cancellationToken) + .ConfigureAwait(false); + wakeSignal.Notify(); + logger.ConcurrencyGroupRateLimitOverrideCleared(groupKey); + } + + public ValueTask GetRateLimitOverrideAsync( + string groupKey, + CancellationToken cancellationToken = default) + { + SubmissionValidator.ValidateConcurrencyGroupKey(groupKey); + + return store.GetConcurrencyRateLimitOverrideAsync(groupKey, cancellationToken); + } + private static void ValidateLimit(int limit) { if (limit <= 0) @@ -60,4 +146,33 @@ private static void ValidateLimit(int limit) throw new ArgumentOutOfRangeException(nameof(limit), limit, "Concurrency group limits must be positive."); } } + + private static void ValidateRateLimit(ConcurrencyGroupRateLimit rateLimit) + { + ArgumentNullException.ThrowIfNull(rateLimit); + + if (rateLimit.PermitCount <= 0) + { + throw new ArgumentOutOfRangeException( + nameof(rateLimit), + rateLimit.PermitCount, + "Concurrency group rate-limit permit counts must be positive."); + } + + if (rateLimit.Period <= TimeSpan.Zero) + { + throw new ArgumentOutOfRangeException( + nameof(rateLimit), + rateLimit.Period, + "Concurrency group rate-limit periods must be positive."); + } + + if (rateLimit.Period.TotalMicroseconds / rateLimit.PermitCount < 1) + { + throw new ArgumentOutOfRangeException( + nameof(rateLimit), + rateLimit, + "Concurrency group rate-limit emission intervals must be at least one microsecond."); + } + } } diff --git a/src/Sheddueller/Storage/ClaimJobResult.cs b/src/Sheddueller/Storage/ClaimJobResult.cs index ffcacbe..4773806 100644 --- a/src/Sheddueller/Storage/ClaimJobResult.cs +++ b/src/Sheddueller/Storage/ClaimJobResult.cs @@ -15,5 +15,6 @@ public sealed record Claimed(ClaimedJob Job) : ClaimJobResult; /// /// No job is currently claimable. /// - public sealed record NoJobAvailable : ClaimJobResult; + /// The earliest known time at which rate-limited work may become claimable. + public sealed record NoJobAvailable(DateTimeOffset? NextClaimAtUtc = null) : ClaimJobResult; } diff --git a/src/Sheddueller/Storage/ClearConcurrencyDefaultRateLimitRequest.cs b/src/Sheddueller/Storage/ClearConcurrencyDefaultRateLimitRequest.cs new file mode 100644 index 0000000..aad466d --- /dev/null +++ b/src/Sheddueller/Storage/ClearConcurrencyDefaultRateLimitRequest.cs @@ -0,0 +1,8 @@ +namespace Sheddueller.Storage; + +/// +/// Store request for clearing a concurrency-group code-defined default rate limit. +/// +public sealed record ClearConcurrencyDefaultRateLimitRequest( + string GroupKey, + DateTimeOffset UpdatedAtUtc); diff --git a/src/Sheddueller/Storage/ClearConcurrencyRateLimitOverrideRequest.cs b/src/Sheddueller/Storage/ClearConcurrencyRateLimitOverrideRequest.cs new file mode 100644 index 0000000..a3a2288 --- /dev/null +++ b/src/Sheddueller/Storage/ClearConcurrencyRateLimitOverrideRequest.cs @@ -0,0 +1,8 @@ +namespace Sheddueller.Storage; + +/// +/// Store request for clearing a concurrency-group live rate-limit override. +/// +public sealed record ClearConcurrencyRateLimitOverrideRequest( + string GroupKey, + DateTimeOffset UpdatedAtUtc); diff --git a/src/Sheddueller/Storage/IJobStore.cs b/src/Sheddueller/Storage/IJobStore.cs index fb3a1f4..b0105a3 100644 --- a/src/Sheddueller/Storage/IJobStore.cs +++ b/src/Sheddueller/Storage/IJobStore.cs @@ -124,6 +124,54 @@ ValueTask ClearConcurrencyLimitOverrideAsync( string groupKey, CancellationToken cancellationToken = default); + /// + /// Persists a limited live concurrency-group rate override. + /// + ValueTask SetConcurrencyRateLimitAsync( + SetConcurrencyRateLimitRequest request, + CancellationToken cancellationToken = default) + => throw new NotSupportedException("This job store does not support concurrency-group rate limits."); + + /// + /// Persists a code-defined concurrency-group default rate without clearing a live override. + /// + ValueTask SetConcurrencyDefaultRateLimitAsync( + SetConcurrencyDefaultRateLimitRequest request, + CancellationToken cancellationToken = default) + => throw new NotSupportedException("This job store does not support concurrency-group rate limits."); + + /// + /// Clears a code-defined concurrency-group default rate without clearing a live override. + /// + ValueTask ClearConcurrencyDefaultRateLimitAsync( + ClearConcurrencyDefaultRateLimitRequest request, + CancellationToken cancellationToken = default) + => throw new NotSupportedException("This job store does not support concurrency-group rate limits."); + + /// + /// Persists an explicitly unlimited live concurrency-group rate override. + /// + ValueTask SetConcurrencyUnlimitedRateLimitAsync( + SetConcurrencyUnlimitedRateLimitRequest request, + CancellationToken cancellationToken = default) + => throw new NotSupportedException("This job store does not support concurrency-group rate limits."); + + /// + /// Clears a live concurrency-group rate-limit override. + /// + ValueTask ClearConcurrencyRateLimitOverrideAsync( + ClearConcurrencyRateLimitOverrideRequest request, + CancellationToken cancellationToken = default) + => throw new NotSupportedException("This job store does not support concurrency-group rate limits."); + + /// + /// Gets the live concurrency-group rate-limit override. + /// + ValueTask GetConcurrencyRateLimitOverrideAsync( + string groupKey, + CancellationToken cancellationToken = default) + => ValueTask.FromResult(new ConcurrencyGroupRateLimitOverride(ConcurrencyGroupRateLimitOverrideKind.Inherit)); + /// /// Creates or updates a recurring schedule definition. /// diff --git a/src/Sheddueller/Storage/SetConcurrencyDefaultRateLimitRequest.cs b/src/Sheddueller/Storage/SetConcurrencyDefaultRateLimitRequest.cs new file mode 100644 index 0000000..c59f6fe --- /dev/null +++ b/src/Sheddueller/Storage/SetConcurrencyDefaultRateLimitRequest.cs @@ -0,0 +1,9 @@ +namespace Sheddueller.Storage; + +/// +/// Store request for setting a concurrency-group code-defined default rate limit. +/// +public sealed record SetConcurrencyDefaultRateLimitRequest( + string GroupKey, + ConcurrencyGroupRateLimit RateLimit, + DateTimeOffset UpdatedAtUtc); diff --git a/src/Sheddueller/Storage/SetConcurrencyRateLimitRequest.cs b/src/Sheddueller/Storage/SetConcurrencyRateLimitRequest.cs new file mode 100644 index 0000000..8c72bfa --- /dev/null +++ b/src/Sheddueller/Storage/SetConcurrencyRateLimitRequest.cs @@ -0,0 +1,9 @@ +namespace Sheddueller.Storage; + +/// +/// Store request for setting a concurrency-group live rate-limit override. +/// +public sealed record SetConcurrencyRateLimitRequest( + string GroupKey, + ConcurrencyGroupRateLimit RateLimit, + DateTimeOffset UpdatedAtUtc); diff --git a/src/Sheddueller/Storage/SetConcurrencyUnlimitedRateLimitRequest.cs b/src/Sheddueller/Storage/SetConcurrencyUnlimitedRateLimitRequest.cs new file mode 100644 index 0000000..ad29fd5 --- /dev/null +++ b/src/Sheddueller/Storage/SetConcurrencyUnlimitedRateLimitRequest.cs @@ -0,0 +1,8 @@ +namespace Sheddueller.Storage; + +/// +/// Store request for setting an explicitly unlimited concurrency-group live rate override. +/// +public sealed record SetConcurrencyUnlimitedRateLimitRequest( + string GroupKey, + DateTimeOffset UpdatedAtUtc); diff --git a/test/Sheddueller.Dashboard.Tests/DashboardEndpointTests.cs b/test/Sheddueller.Dashboard.Tests/DashboardEndpointTests.cs index 0af2340..0b43772 100644 --- a/test/Sheddueller.Dashboard.Tests/DashboardEndpointTests.cs +++ b/test/Sheddueller.Dashboard.Tests/DashboardEndpointTests.cs @@ -270,6 +270,7 @@ public async Task ConcurrencyGroups_KnownData_RendersRegistry() html.ShouldContain("sd-table-switch"); html.ShouldContain("Filter by group key"); html.ShouldContain("Saturated only"); + html.ShouldContain("Rate-limited only"); html.ShouldContain("Has blocked jobs"); html.ShouldContain("type=\"checkbox\""); AssertAppearsBefore(html, "Saturated only", "aria-label=\"Filter by group key\""); @@ -279,6 +280,9 @@ public async Task ConcurrencyGroups_KnownData_RendersRegistry() html.ShouldContain("bg_maintenance"); html.ShouldContain("db_vacuum_ops"); html.ShouldContain("Limit Source"); + html.ShouldContain("Effective Rate"); + html.ShouldContain("Rate Source"); + html.ShouldContain("Next Permit"); html.ShouldContain("Override"); html.ShouldContain("Code default"); html.ShouldContain("Built-in default"); @@ -288,6 +292,7 @@ public async Task ConcurrencyGroups_KnownData_RendersRegistry() html.ShouldContain("High Load"); html.ShouldContain("Nominal"); html.ShouldContain("Blocked Work"); + html.ShouldContain("Rate Limited"); html.ShouldContain("2026-04-20 12:02:01 UTC"); html.ShouldContain("Load More Records"); html.ShouldContain("Showing 1-4 of 4 groups with more available"); @@ -303,8 +308,13 @@ public async Task ConcurrencyGroups_WithManager_RendersLimitEditActions() html.ShouldContain("Actions"); html.ShouldContain("Edit Limit"); html.ShouldContain("Reset Override"); + html.ShouldContain("Edit Rate Limit"); + html.ShouldContain("Set Unlimited Rate"); + html.ShouldContain("Reset Rate Override"); html.ShouldContain("aria-label=\"Edit limit for concurrency group pool_etl_heavy\""); html.ShouldContain("aria-label=\"Reset limit override for concurrency group pool_etl_heavy\""); + html.ShouldContain("aria-label=\"Edit rate limit for concurrency group pool_etl_heavy\""); + html.ShouldContain("aria-label=\"Set unlimited rate override for concurrency group pool_etl_heavy\""); } [Fact] @@ -1296,16 +1306,21 @@ private sealed class StubConcurrencyGroupInspectionReader : IConcurrencyGroupIns "api_sync_workers", EffectiveLimit: 100, CurrentOccupancy: 85, - BlockedJobCount: 0, + BlockedJobCount: 5, IsSaturated: false, UpdatedAtUtc.AddMinutes(-1)) { DefaultLimit = 100, + DefaultRateLimit = new ConcurrencyGroupRateLimit(2, TimeSpan.FromSeconds(1)), + EffectiveRateLimit = new ConcurrencyGroupRateLimit(2, TimeSpan.FromSeconds(1)), + NextRatePermitAtUtc = UpdatedAtUtc.AddSeconds(1), + IsRateLimited = true, + RateBlockedJobCount = 5, }, new( "bg_maintenance", - EffectiveLimit: 1, - CurrentOccupancy: 0, + EffectiveLimit: 10, + CurrentOccupancy: 8, BlockedJobCount: 0, IsSaturated: false, UpdatedAtUtc.AddMinutes(-5)), @@ -1325,6 +1340,7 @@ public ValueTask SearchConcurrencyGroupsAsync( var filtered = Groups .Where(group => query.GroupKey is null || string.Equals(group.GroupKey, query.GroupKey, StringComparison.Ordinal)) .Where(group => query.IsSaturated is null || group.IsSaturated == query.IsSaturated.Value) + .Where(group => query.IsRateLimited is null || group.IsRateLimited == query.IsRateLimited.Value) .Where(group => query.HasBlockedJobs is null || (group.BlockedJobCount > 0) == query.HasBlockedJobs.Value) .ToArray(); var groups = filtered diff --git a/test/Sheddueller.Dashboard.Tests/DashboardFilterTests.cs b/test/Sheddueller.Dashboard.Tests/DashboardFilterTests.cs index 80bf300..af3eea6 100644 --- a/test/Sheddueller.Dashboard.Tests/DashboardFilterTests.cs +++ b/test/Sheddueller.Dashboard.Tests/DashboardFilterTests.cs @@ -150,6 +150,7 @@ public void ConcurrencyGroupFilters_QueryAndClientFilter_MapSwitchesAndSubstring GroupKey = "API", SaturatedOnly = true, HasBlockedJobsOnly = true, + RateLimitedOnly = true, }; var groups = new[] { @@ -164,6 +165,7 @@ public void ConcurrencyGroupFilters_QueryAndClientFilter_MapSwitchesAndSubstring query.GroupKey.ShouldBeNull(); query.IsSaturated.ShouldBe(true); query.HasBlockedJobs.ShouldBe(true); + query.IsRateLimited.ShouldBe(true); query.PageSize.ShouldBe(10); } diff --git a/test/Sheddueller.Postgres.Tests/Operations/SetConcurrencyRateLimitOperationTests.cs b/test/Sheddueller.Postgres.Tests/Operations/SetConcurrencyRateLimitOperationTests.cs new file mode 100644 index 0000000..3684793 --- /dev/null +++ b/test/Sheddueller.Postgres.Tests/Operations/SetConcurrencyRateLimitOperationTests.cs @@ -0,0 +1,88 @@ +namespace Sheddueller.Postgres.Tests.Operations; + +using Sheddueller.Storage; + +using Shouldly; + +public sealed class SetConcurrencyRateLimitOperationTests(PostgresFixture fixture) : IClassFixture +{ + private static readonly ConcurrencyGroupRateLimit DefaultRate = new(2, TimeSpan.FromSeconds(1)); + private static readonly ConcurrencyGroupRateLimit OverrideRate = new(3, TimeSpan.FromSeconds(2)); + + [Fact] + public async Task RateLimit_DefaultOverrideUnlimitedAndClear_UsesEffectivePrecedence() + { + await using var context = await PostgresTestContext.CreateMigratedAsync(fixture); + + await context.Store.SetConcurrencyDefaultRateLimitAsync( + new SetConcurrencyDefaultRateLimitRequest("shared", DefaultRate, DateTimeOffset.UtcNow)); + var defaultOnly = (await context.ReadConcurrencyGroupAsync("shared")).ShouldNotBeNull(); + defaultOnly.EffectiveRateLimit.ShouldBe(DefaultRate); + (await context.Store.GetConcurrencyRateLimitOverrideAsync("shared")).Kind + .ShouldBe(ConcurrencyGroupRateLimitOverrideKind.Inherit); + + await context.Store.SetConcurrencyRateLimitAsync( + new SetConcurrencyRateLimitRequest("shared", OverrideRate, DateTimeOffset.UtcNow)); + var overridden = (await context.ReadConcurrencyGroupAsync("shared")).ShouldNotBeNull(); + overridden.EffectiveRateLimit.ShouldBe(OverrideRate); + (await context.Store.GetConcurrencyRateLimitOverrideAsync("shared")) + .ShouldBe(new ConcurrencyGroupRateLimitOverride( + ConcurrencyGroupRateLimitOverrideKind.Limited, + OverrideRate)); + + await context.Store.SetConcurrencyUnlimitedRateLimitAsync( + new SetConcurrencyUnlimitedRateLimitRequest("shared", DateTimeOffset.UtcNow)); + var unlimited = (await context.ReadConcurrencyGroupAsync("shared")).ShouldNotBeNull(); + unlimited.RateLimitOverrideEnabled.ShouldBeTrue(); + unlimited.EffectiveRateLimit.ShouldBeNull(); + (await context.Store.GetConcurrencyRateLimitOverrideAsync("shared")).Kind + .ShouldBe(ConcurrencyGroupRateLimitOverrideKind.Unlimited); + + await context.Store.ClearConcurrencyRateLimitOverrideAsync( + new ClearConcurrencyRateLimitOverrideRequest("shared", DateTimeOffset.UtcNow)); + var inherited = (await context.ReadConcurrencyGroupAsync("shared")).ShouldNotBeNull(); + inherited.RateLimitOverrideEnabled.ShouldBeFalse(); + inherited.EffectiveRateLimit.ShouldBe(DefaultRate); + + await context.Store.ClearConcurrencyDefaultRateLimitAsync( + new ClearConcurrencyDefaultRateLimitRequest("shared", DateTimeOffset.UtcNow)); + (await context.ReadConcurrencyGroupAsync("shared")).ShouldNotBeNull().EffectiveRateLimit.ShouldBeNull(); + } + + [Fact] + public async Task RateLimit_UnchangedDefault_PreservesTimingState() + { + await using var context = await PostgresTestContext.CreateMigratedAsync(fixture); + var jobId = Guid.NewGuid(); + + await context.Store.SetConcurrencyDefaultRateLimitAsync( + new SetConcurrencyDefaultRateLimitRequest("shared", DefaultRate, DateTimeOffset.UtcNow)); + await context.Store.EnqueueAsync(PostgresTestData.CreateRequest(jobId, groupKeys: ["shared"])); + await PostgresTestData.ClaimAsync(context.Store); + var before = (await context.ReadConcurrencyGroupAsync("shared")).ShouldNotBeNull().RateTheoreticalArrivalAtUtc; + before.ShouldNotBeNull(); + + await context.Store.SetConcurrencyDefaultRateLimitAsync( + new SetConcurrencyDefaultRateLimitRequest("shared", DefaultRate, DateTimeOffset.UtcNow)); + + (await context.ReadConcurrencyGroupAsync("shared")).ShouldNotBeNull().RateTheoreticalArrivalAtUtc.ShouldBe(before); + } + + [Fact] + public async Task RateLimit_EffectiveChange_ResetsTimingState() + { + await using var context = await PostgresTestContext.CreateMigratedAsync(fixture); + var jobId = Guid.NewGuid(); + + await context.Store.SetConcurrencyDefaultRateLimitAsync( + new SetConcurrencyDefaultRateLimitRequest("shared", DefaultRate, DateTimeOffset.UtcNow)); + await context.Store.EnqueueAsync(PostgresTestData.CreateRequest(jobId, groupKeys: ["shared"])); + await PostgresTestData.ClaimAsync(context.Store); + (await context.ReadConcurrencyGroupAsync("shared")).ShouldNotBeNull().RateTheoreticalArrivalAtUtc.ShouldNotBeNull(); + + await context.Store.SetConcurrencyDefaultRateLimitAsync( + new SetConcurrencyDefaultRateLimitRequest("shared", OverrideRate, DateTimeOffset.UtcNow)); + + (await context.ReadConcurrencyGroupAsync("shared")).ShouldNotBeNull().RateTheoreticalArrivalAtUtc.ShouldBeNull(); + } +} diff --git a/test/Sheddueller.Postgres.Tests/Operations/TryClaimNextJobOperationTests.cs b/test/Sheddueller.Postgres.Tests/Operations/TryClaimNextJobOperationTests.cs index f4cdf02..c2e8f0e 100644 --- a/test/Sheddueller.Postgres.Tests/Operations/TryClaimNextJobOperationTests.cs +++ b/test/Sheddueller.Postgres.Tests/Operations/TryClaimNextJobOperationTests.cs @@ -120,4 +120,142 @@ public async Task TryClaim_ConcurrentNodes_ClaimsJobOnlyOnce() results.Count(result => result is ClaimJobResult.Claimed).ShouldBe(1); results.Count(result => result is ClaimJobResult.NoJobAvailable).ShouldBe(1); } + + [Fact] + public async Task TryClaim_SmoothRateLimit_SpacesClaimsAndReturnsNextClaimTime() + { + await using var context = await PostgresTestContext.CreateMigratedAsync(fixture); + await ConfigureRateAsync(context, "rate", new ConcurrencyGroupRateLimit(2, TimeSpan.FromSeconds(1))); + await context.Store.EnqueueAsync(PostgresTestData.CreateRequest(Guid.NewGuid(), groupKeys: ["rate"])); + await context.Store.EnqueueAsync(PostgresTestData.CreateRequest(Guid.NewGuid(), groupKeys: ["rate"])); + + await PostgresTestData.ClaimAsync(context.Store); + var blockedAt = DateTimeOffset.UtcNow; + var unavailable = (await context.Store.TryClaimNextAsync(PostgresTestData.ClaimRequest())) + .ShouldBeOfType(); + + unavailable.NextClaimAtUtc.ShouldNotBeNull().ShouldBeGreaterThan(blockedAt); + await Task.Delay(TimeSpan.FromMilliseconds(550)); + _ = await PostgresTestData.ClaimAsync(context.Store); + } + + [Fact] + public async Task TryClaim_SmoothRateLimit_IdleTimeDoesNotCreateBurstCredit() + { + await using var context = await PostgresTestContext.CreateMigratedAsync(fixture); + await ConfigureRateAsync(context, "rate", new ConcurrencyGroupRateLimit(2, TimeSpan.FromSeconds(1))); + await context.Store.EnqueueAsync(PostgresTestData.CreateRequest(Guid.NewGuid(), groupKeys: ["rate"])); + await context.Store.EnqueueAsync(PostgresTestData.CreateRequest(Guid.NewGuid(), groupKeys: ["rate"])); + await Task.Delay(TimeSpan.FromMilliseconds(600)); + + await PostgresTestData.ClaimAsync(context.Store); + + (await context.Store.TryClaimNextAsync(PostgresTestData.ClaimRequest())) + .ShouldBeOfType() + .NextClaimAtUtc.ShouldNotBeNull(); + } + + [Fact] + public async Task TryClaim_RateBlockedHighPriority_ClaimsLaterEligibleJob() + { + await using var context = await PostgresTestContext.CreateMigratedAsync(fixture); + await ConfigureRateAsync(context, "rate", new ConcurrencyGroupRateLimit(1, TimeSpan.FromSeconds(1))); + await context.Store.EnqueueAsync(PostgresTestData.CreateRequest(Guid.NewGuid(), priority: 200, groupKeys: ["rate"])); + await PostgresTestData.ClaimAsync(context.Store); + var blocked = Guid.NewGuid(); + var eligible = Guid.NewGuid(); + await context.Store.EnqueueAsync(PostgresTestData.CreateRequest(blocked, priority: 100, groupKeys: ["rate"])); + await context.Store.EnqueueAsync(PostgresTestData.CreateRequest(eligible, priority: 0)); + + (await PostgresTestData.ClaimAsync(context.Store)).JobId.ShouldBe(eligible); + (await context.ReadJobAsync(blocked)).State.ShouldBe("Queued"); + } + + [Fact] + public async Task TryClaim_MultipleGroups_RateBlockedGroupDoesNotConsumeOtherGroup() + { + await using var context = await PostgresTestContext.CreateMigratedAsync(fixture); + var rate = new ConcurrencyGroupRateLimit(1, TimeSpan.FromSeconds(1)); + await ConfigureRateAsync(context, "rate-a", rate); + await ConfigureRateAsync(context, "rate-b", rate); + await context.Store.EnqueueAsync(PostgresTestData.CreateRequest(Guid.NewGuid(), groupKeys: ["rate-a"])); + await PostgresTestData.ClaimAsync(context.Store); + await context.Store.EnqueueAsync(PostgresTestData.CreateRequest(Guid.NewGuid(), priority: 100, groupKeys: ["rate-a", "rate-b"])); + var bOnly = Guid.NewGuid(); + await context.Store.EnqueueAsync(PostgresTestData.CreateRequest(bOnly, groupKeys: ["rate-b"])); + + (await PostgresTestData.ClaimAsync(context.Store)).JobId.ShouldBe(bOnly); + } + + [Fact] + public async Task TryClaim_ConcurrencyBlockedGroup_DoesNotConsumeRatePermit() + { + await using var context = await PostgresTestContext.CreateMigratedAsync(fixture); + await ConfigureRateAsync(context, "rate", new ConcurrencyGroupRateLimit(1, TimeSpan.FromSeconds(1))); + var holder = Guid.NewGuid(); + await context.Store.EnqueueAsync(PostgresTestData.CreateRequest(holder, groupKeys: ["capacity"])); + var holderClaim = await PostgresTestData.ClaimAsync(context.Store); + var blocked = Guid.NewGuid(); + await context.Store.EnqueueAsync(PostgresTestData.CreateRequest(blocked, groupKeys: ["capacity", "rate"])); + + (await context.Store.TryClaimNextAsync(PostgresTestData.ClaimRequest())) + .ShouldBeOfType(); + (await context.ReadConcurrencyGroupAsync("rate")).ShouldNotBeNull().RateTheoreticalArrivalAtUtc.ShouldBeNull(); + + await context.Store.MarkCompletedAsync( + new CompleteJobRequest(holder, "node-1", holderClaim.LeaseToken, DateTimeOffset.UtcNow)); + (await PostgresTestData.ClaimAsync(context.Store)).JobId.ShouldBe(blocked); + } + + [Fact] + public async Task TryClaim_ConcurrentNodes_ConsumeOnlyOneRatePermit() + { + await using var context = await PostgresTestContext.CreateMigratedAsync(fixture); + await ConfigureRateAsync(context, "rate", new ConcurrencyGroupRateLimit(1, TimeSpan.FromMilliseconds(300))); + await context.Store.EnqueueAsync(PostgresTestData.CreateRequest(Guid.NewGuid(), groupKeys: ["rate"])); + await context.Store.EnqueueAsync(PostgresTestData.CreateRequest(Guid.NewGuid(), groupKeys: ["rate"])); + + var results = await Task.WhenAll( + context.Store.TryClaimNextAsync(PostgresTestData.ClaimRequest("node-a")).AsTask(), + context.Store.TryClaimNextAsync(PostgresTestData.ClaimRequest("node-b")).AsTask()); + + results.Count(result => result is ClaimJobResult.Claimed).ShouldBe(1); + results.Count(result => result is ClaimJobResult.NoJobAvailable).ShouldBe(1); + } + + [Fact] + public async Task TryClaim_RetryAttempt_ConsumesAnotherRatePermit() + { + await using var context = await PostgresTestContext.CreateMigratedAsync(fixture); + await ConfigureRateAsync(context, "rate", new ConcurrencyGroupRateLimit(1, TimeSpan.FromMilliseconds(300))); + var jobId = Guid.NewGuid(); + await context.Store.EnqueueAsync(PostgresTestData.CreateRequest( + jobId, + maxAttempts: 2, + retryBackoffKind: RetryBackoffKind.Fixed, + retryBaseDelay: TimeSpan.FromMilliseconds(1), + groupKeys: ["rate"])); + var first = await PostgresTestData.ClaimAsync(context.Store); + await context.Store.MarkFailedAsync( + new FailJobRequest(jobId, "node-1", first.LeaseToken, DateTimeOffset.UtcNow, PostgresTestData.CreateFailure())); + await Task.Delay(TimeSpan.FromMilliseconds(20)); + + (await context.Store.TryClaimNextAsync(PostgresTestData.ClaimRequest())) + .ShouldBeOfType() + .NextClaimAtUtc.ShouldNotBeNull(); + + await Task.Delay(TimeSpan.FromMilliseconds(330)); + (await PostgresTestData.ClaimAsync(context.Store)).AttemptCount.ShouldBe(2); + } + + private static async ValueTask ConfigureRateAsync( + PostgresTestContext context, + string groupKey, + ConcurrencyGroupRateLimit rateLimit) + { + await context.Store.SetConcurrencyLimitAsync( + new SetConcurrencyLimitRequest(groupKey, 10, DateTimeOffset.UtcNow)); + await context.Store.SetConcurrencyDefaultRateLimitAsync( + new SetConcurrencyDefaultRateLimitRequest(groupKey, rateLimit, DateTimeOffset.UtcNow)); + } } diff --git a/test/Sheddueller.Postgres.Tests/PostgresMigrationTests.cs b/test/Sheddueller.Postgres.Tests/PostgresMigrationTests.cs index 5ffa1eb..37543a8 100644 --- a/test/Sheddueller.Postgres.Tests/PostgresMigrationTests.cs +++ b/test/Sheddueller.Postgres.Tests/PostgresMigrationTests.cs @@ -238,6 +238,39 @@ await ExecuteAsync( .ShouldBe(1); } + [Fact] + public async Task Migration_FreshSchema_CreatesConcurrencyRateColumns() + { + await using var context = await PostgresTestContext.CreateMigratedAsync(fixture); + + await ExecuteAsync( + context, + $""" + insert into {context.Table("concurrency_groups")} ( + group_key, + configured_limit, + rate_limit_override_enabled, + configured_rate_permit_count, + configured_rate_period, + default_rate_permit_count, + default_rate_period, + in_use_count, + updated_at_utc) + values + ('override', null, true, 5, interval '1 second', 2, interval '1 minute', 0, transaction_timestamp()), + ('unlimited', null, true, null, null, 2, interval '1 minute', 0, transaction_timestamp()), + ('default', null, false, null, null, 3, interval '1 minute', 0, transaction_timestamp()), + ('built-in', null, false, null, null, null, null, 0, transaction_timestamp()); + """); + + (await context.ReadConcurrencyGroupAsync("override")).ShouldNotBeNull().EffectiveRateLimit + .ShouldBe(new ConcurrencyGroupRateLimit(5, TimeSpan.FromSeconds(1))); + (await context.ReadConcurrencyGroupAsync("unlimited")).ShouldNotBeNull().EffectiveRateLimit.ShouldBeNull(); + (await context.ReadConcurrencyGroupAsync("default")).ShouldNotBeNull().EffectiveRateLimit + .ShouldBe(new ConcurrencyGroupRateLimit(3, TimeSpan.FromMinutes(1))); + (await context.ReadConcurrencyGroupAsync("built-in")).ShouldNotBeNull().EffectiveRateLimit.ShouldBeNull(); + } + [Fact] public async Task Migration_FreshSchema_CreatesTagOrdinalColumnsAndIndexes() { diff --git a/test/Sheddueller.Postgres.Tests/PostgresTestContext.cs b/test/Sheddueller.Postgres.Tests/PostgresTestContext.cs index 82934e3..bb035f0 100644 --- a/test/Sheddueller.Postgres.Tests/PostgresTestContext.cs +++ b/test/Sheddueller.Postgres.Tests/PostgresTestContext.cs @@ -313,7 +313,20 @@ public async ValueTask> ReadScheduleGroupKeysAsync(string { await using var command = this.DataSource.CreateCommand( $""" - select group_key, configured_limit, default_limit, effective_limit, in_use_count + select + group_key, + configured_limit, + default_limit, + effective_limit, + in_use_count, + rate_limit_override_enabled, + configured_rate_permit_count, + configured_rate_period, + default_rate_permit_count, + default_rate_period, + effective_rate_permit_count, + effective_rate_period, + rate_theoretical_arrival_at_utc from {this.Table("concurrency_groups")} where group_key = @group_key; """); @@ -330,7 +343,12 @@ public async ValueTask> ReadScheduleGroupKeysAsync(string reader.IsDBNull(1) ? null : reader.GetInt32(1), reader.IsDBNull(2) ? null : reader.GetInt32(2), reader.GetInt32(3), - reader.GetInt32(4)); + reader.GetInt32(4), + reader.GetBoolean(5), + ReadRateLimit(reader, 6, 7), + ReadRateLimit(reader, 8, 9), + ReadRateLimit(reader, 10, 11), + reader.IsDBNull(12) ? null : ToDateTimeOffset(reader.GetValue(12))); } public async ValueTask CountJobsForScheduleAsync(string scheduleKey) @@ -400,6 +418,14 @@ private static DateTimeOffset ToDateTimeOffset(object value) DateTime dateTime => new DateTimeOffset(DateTime.SpecifyKind(dateTime, DateTimeKind.Utc)), _ => throw new InvalidOperationException($"Unexpected timestamp type '{value.GetType()}'."), }; + + private static ConcurrencyGroupRateLimit? ReadRateLimit( + NpgsqlDataReader reader, + int countOrdinal, + int periodOrdinal) + => reader.IsDBNull(countOrdinal) + ? null + : new ConcurrencyGroupRateLimit(reader.GetInt32(countOrdinal), reader.GetTimeSpan(periodOrdinal)); } internal sealed record PostgresJobRow( @@ -464,4 +490,9 @@ internal sealed record PostgresConcurrencyGroupRow( int? ConfiguredLimit, int? DefaultLimit, int EffectiveLimit, - int InUseCount); + int InUseCount, + bool RateLimitOverrideEnabled, + ConcurrencyGroupRateLimit? ConfiguredRateLimit, + ConcurrencyGroupRateLimit? DefaultRateLimit, + ConcurrencyGroupRateLimit? EffectiveRateLimit, + DateTimeOffset? RateTheoreticalArrivalAtUtc); diff --git a/test/Sheddueller.ProviderContracts/InspectionContractTests.cs b/test/Sheddueller.ProviderContracts/InspectionContractTests.cs index c0b5dc7..500182a 100644 --- a/test/Sheddueller.ProviderContracts/InspectionContractTests.cs +++ b/test/Sheddueller.ProviderContracts/InspectionContractTests.cs @@ -625,6 +625,60 @@ public async Task ConcurrencyGroupView_DefaultAndOverrideLimits_AreVisible() summary => summary.EffectiveLimit.ShouldBe(5)); } + [Fact] + public async Task ConcurrencyGroupView_RateLimitedGroup_ShowsRateStateAndBlockedJobs() + { + await using var context = await this.CreateContextAsync(); + var rateLimit = new ConcurrencyGroupRateLimit(1, TimeSpan.FromSeconds(1)); + var running = Guid.NewGuid(); + var blocked = Guid.NewGuid(); + await context.Store.SetConcurrencyLimitAsync(new SetConcurrencyLimitRequest("api", 10, DateTimeOffset.UtcNow)); + await context.Store.SetConcurrencyDefaultRateLimitAsync( + new SetConcurrencyDefaultRateLimitRequest("api", rateLimit, DateTimeOffset.UtcNow)); + await context.Store.EnqueueAsync(CreateRequest(running, groupKeys: ["api"])); + await context.Store.EnqueueAsync(CreateRequest(blocked, groupKeys: ["api"])); + (await ClaimAsync(context.Store)).JobId.ShouldBe(running); + + var page = await context.ConcurrencyGroupReader.SearchConcurrencyGroupsAsync( + new ConcurrencyGroupInspectionQuery(IsRateLimited: true, HasBlockedJobs: true)); + var detail = await context.ConcurrencyGroupReader.GetConcurrencyGroupAsync("api"); + + page.Groups.ShouldHaveSingleItem().GroupKey.ShouldBe("api"); + detail.ShouldNotBeNull(); + detail.Summary.DefaultRateLimit.ShouldBe(rateLimit); + detail.Summary.HasRateLimitOverride.ShouldBeFalse(); + detail.Summary.OverrideRateLimit.ShouldBeNull(); + detail.Summary.EffectiveRateLimit.ShouldBe(rateLimit); + detail.Summary.NextRatePermitAtUtc.ShouldNotBeNull(); + detail.Summary.IsRateLimited.ShouldBeTrue(); + detail.Summary.ConcurrencyBlockedJobCount.ShouldBe(0); + detail.Summary.RateBlockedJobCount.ShouldBe(1); + detail.Summary.BlockedJobCount.ShouldBe(1); + detail.ConcurrencyBlockedJobIds.ShouldBeEmpty(); + detail.RateBlockedJobIds.ShouldBe([blocked]); + detail.BlockedJobIds.ShouldBe([blocked]); + } + + [Fact] + public async Task ConcurrencyGroupView_UnlimitedOverride_IsVisible() + { + await using var context = await this.CreateContextAsync(); + var defaultRate = new ConcurrencyGroupRateLimit(2, TimeSpan.FromSeconds(1)); + await context.Store.SetConcurrencyDefaultRateLimitAsync( + new SetConcurrencyDefaultRateLimitRequest("api", defaultRate, DateTimeOffset.UtcNow)); + await context.Store.SetConcurrencyUnlimitedRateLimitAsync( + new SetConcurrencyUnlimitedRateLimitRequest("api", DateTimeOffset.UtcNow)); + + var detail = await context.ConcurrencyGroupReader.GetConcurrencyGroupAsync("api"); + + detail.ShouldNotBeNull().Summary.ShouldSatisfyAllConditions( + summary => summary.DefaultRateLimit.ShouldBe(defaultRate), + summary => summary.HasRateLimitOverride.ShouldBeTrue(), + summary => summary.OverrideRateLimit.ShouldBeNull(), + summary => summary.EffectiveRateLimit.ShouldBeNull(), + summary => summary.IsRateLimited.ShouldBeFalse()); + } + [Fact] public async Task ConcurrencyGroupSearch_GroupKeyFilter_IsExactAndReturnsTotalCount() { diff --git a/test/Sheddueller.Tests/ConcurrencyGroupManagerTests.cs b/test/Sheddueller.Tests/ConcurrencyGroupManagerTests.cs index 18f4367..f6bdc0d 100644 --- a/test/Sheddueller.Tests/ConcurrencyGroupManagerTests.cs +++ b/test/Sheddueller.Tests/ConcurrencyGroupManagerTests.cs @@ -61,6 +61,89 @@ public async Task SetLimit_NonPositiveLimit_DoesNotPersist() store.ConcurrencyLimitRequests.ShouldBeEmpty(); } + [Fact] + public async Task SetRateLimit_ValidRate_PersistsOverride() + { + using var provider = CreateProvider(); + var store = provider.GetRequiredService(); + var manager = provider.GetRequiredService(); + var rateLimit = new ConcurrencyGroupRateLimit(2, TimeSpan.FromSeconds(1)); + + await manager.SetRateLimitAsync("api", rateLimit); + + store.ConcurrencyRateLimitRequests.ShouldHaveSingleItem().ShouldSatisfyAllConditions( + request => request.GroupKey.ShouldBe("api"), + request => request.RateLimit.ShouldBe(rateLimit)); + } + + [Fact] + public async Task SetDefaultRateLimit_ValidRate_PersistsDefault() + { + using var provider = CreateProvider(); + var store = provider.GetRequiredService(); + var manager = provider.GetRequiredService(); + var rateLimit = new ConcurrencyGroupRateLimit(10, TimeSpan.FromMinutes(1)); + + await manager.SetDefaultRateLimitAsync("api", rateLimit); + + store.ConcurrencyDefaultRateLimitRequests.ShouldHaveSingleItem().RateLimit.ShouldBe(rateLimit); + store.ConcurrencyRateLimitRequests.ShouldBeEmpty(); + } + + [Fact] + public async Task ClearDefaultRateLimit_ValidGroup_PersistsClear() + { + using var provider = CreateProvider(); + var store = provider.GetRequiredService(); + var manager = provider.GetRequiredService(); + + await manager.ClearDefaultRateLimitAsync("api"); + + store.ClearConcurrencyDefaultRateLimitRequests.ShouldHaveSingleItem().GroupKey.ShouldBe("api"); + } + + [Fact] + public async Task SetUnlimitedRateLimit_ValidGroup_PersistsUnlimitedOverride() + { + using var provider = CreateProvider(); + var store = provider.GetRequiredService(); + var manager = provider.GetRequiredService(); + + await manager.SetUnlimitedRateLimitAsync("api"); + + store.ConcurrencyUnlimitedRateLimitRequests.ShouldHaveSingleItem().GroupKey.ShouldBe("api"); + } + + [Fact] + public async Task ClearRateLimitOverride_ValidGroup_PersistsClear() + { + using var provider = CreateProvider(); + var store = provider.GetRequiredService(); + var manager = provider.GetRequiredService(); + + await manager.ClearRateLimitOverrideAsync("api"); + + store.ClearConcurrencyRateLimitOverrideRequests.ShouldHaveSingleItem().GroupKey.ShouldBe("api"); + } + + [Theory] + [InlineData(0, 1)] + [InlineData(1, 0)] + [InlineData(2, 1)] + public async Task SetRateLimit_InvalidRate_DoesNotPersist( + int permitCount, + long periodTicks) + { + using var provider = CreateProvider(); + var store = provider.GetRequiredService(); + var manager = provider.GetRequiredService(); + var rateLimit = new ConcurrencyGroupRateLimit(permitCount, TimeSpan.FromTicks(periodTicks)); + + await Should.ThrowAsync(async () => await manager.SetRateLimitAsync("api", rateLimit)); + + store.ConcurrencyRateLimitRequests.ShouldBeEmpty(); + } + private static ServiceProvider CreateProvider() { var services = new ServiceCollection(); diff --git a/test/Sheddueller.Tests/RecordingJobStore.cs b/test/Sheddueller.Tests/RecordingJobStore.cs index b385ac5..99d75e7 100644 --- a/test/Sheddueller.Tests/RecordingJobStore.cs +++ b/test/Sheddueller.Tests/RecordingJobStore.cs @@ -11,6 +11,11 @@ internal sealed class RecordingJobStore : IJobStore private readonly List concurrencyLimitRequests = []; private readonly List concurrencyDefaultLimitRequests = []; private readonly List clearConcurrencyLimitOverrideRequests = []; + private readonly List concurrencyRateLimitRequests = []; + private readonly List concurrencyDefaultRateLimitRequests = []; + private readonly List clearConcurrencyDefaultRateLimitRequests = []; + private readonly List concurrencyUnlimitedRateLimitRequests = []; + private readonly List clearConcurrencyRateLimitOverrideRequests = []; private long nextSequence; public IReadOnlyList EnqueuedRequests => this.enqueuedRequests; @@ -27,6 +32,16 @@ internal sealed class RecordingJobStore : IJobStore public IReadOnlyList ClearConcurrencyLimitOverrideRequests => this.clearConcurrencyLimitOverrideRequests; + public IReadOnlyList ConcurrencyRateLimitRequests => this.concurrencyRateLimitRequests; + + public IReadOnlyList ConcurrencyDefaultRateLimitRequests => this.concurrencyDefaultRateLimitRequests; + + public IReadOnlyList ClearConcurrencyDefaultRateLimitRequests => this.clearConcurrencyDefaultRateLimitRequests; + + public IReadOnlyList ConcurrencyUnlimitedRateLimitRequests => this.concurrencyUnlimitedRateLimitRequests; + + public IReadOnlyList ClearConcurrencyRateLimitOverrideRequests => this.clearConcurrencyRateLimitOverrideRequests; + public RecurringScheduleUpsertResult CreateOrUpdateRecurringScheduleResult { get; set; } = RecurringScheduleUpsertResult.Created; public RecurringScheduleTriggerResult TriggerResult { get; set; } = new(RecurringScheduleTriggerStatus.NotFound); @@ -159,6 +174,56 @@ public ValueTask ClearConcurrencyLimitOverrideAsync( CancellationToken cancellationToken = default) => throw CreateUnsupportedException(); + public ValueTask SetConcurrencyRateLimitAsync( + SetConcurrencyRateLimitRequest request, + CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + this.concurrencyRateLimitRequests.Add(request); + return ValueTask.CompletedTask; + } + + public ValueTask SetConcurrencyDefaultRateLimitAsync( + SetConcurrencyDefaultRateLimitRequest request, + CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + this.concurrencyDefaultRateLimitRequests.Add(request); + return ValueTask.CompletedTask; + } + + public ValueTask ClearConcurrencyDefaultRateLimitAsync( + ClearConcurrencyDefaultRateLimitRequest request, + CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + this.clearConcurrencyDefaultRateLimitRequests.Add(request); + return ValueTask.CompletedTask; + } + + public ValueTask SetConcurrencyUnlimitedRateLimitAsync( + SetConcurrencyUnlimitedRateLimitRequest request, + CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + this.concurrencyUnlimitedRateLimitRequests.Add(request); + return ValueTask.CompletedTask; + } + + public ValueTask ClearConcurrencyRateLimitOverrideAsync( + ClearConcurrencyRateLimitOverrideRequest request, + CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + this.clearConcurrencyRateLimitOverrideRequests.Add(request); + return ValueTask.CompletedTask; + } + + public ValueTask GetConcurrencyRateLimitOverrideAsync( + string groupKey, + CancellationToken cancellationToken = default) + => ValueTask.FromResult(new ConcurrencyGroupRateLimitOverride(ConcurrencyGroupRateLimitOverrideKind.Inherit)); + public ValueTask CreateOrUpdateRecurringScheduleAsync( UpsertRecurringScheduleRequest request, CancellationToken cancellationToken = default) diff --git a/test/Sheddueller.Worker.Tests/WorkerRateLimitTimingTests.cs b/test/Sheddueller.Worker.Tests/WorkerRateLimitTimingTests.cs new file mode 100644 index 0000000..e656fb2 --- /dev/null +++ b/test/Sheddueller.Worker.Tests/WorkerRateLimitTimingTests.cs @@ -0,0 +1,42 @@ +namespace Sheddueller.Worker.Tests; + +using Sheddueller.Worker.Internal; + +using Shouldly; + +public sealed class WorkerRateLimitTimingTests +{ + private static readonly DateTimeOffset Now = new(2026, 7, 24, 12, 0, 0, TimeSpan.Zero); + + [Fact] + public void WaitTimeout_RatePermitBeforePollingDeadline_UsesRatePermit() + => ShedduellerWorker.CalculateWaitTimeout( + TimeSpan.FromSeconds(1), + Now, + Now.AddMilliseconds(250)) + .ShouldBe(TimeSpan.FromMilliseconds(250)); + + [Fact] + public void WaitTimeout_RatePermitAfterPollingDeadline_UsesPollingInterval() + => ShedduellerWorker.CalculateWaitTimeout( + TimeSpan.FromSeconds(1), + Now, + Now.AddSeconds(5)) + .ShouldBe(TimeSpan.FromSeconds(1)); + + [Fact] + public void WaitTimeout_RatePermitAlreadyDue_ReturnsZero() + => ShedduellerWorker.CalculateWaitTimeout( + TimeSpan.FromSeconds(1), + Now, + Now) + .ShouldBe(TimeSpan.Zero); + + [Fact] + public void WaitTimeout_NoRatePermit_UsesPollingInterval() + => ShedduellerWorker.CalculateWaitTimeout( + TimeSpan.FromSeconds(1), + Now, + nextClaimAtUtc: null) + .ShouldBe(TimeSpan.FromSeconds(1)); +} From 27861c84e86c07373d837777bf641ed49a0c3c7b Mon Sep 17 00:00:00 2001 From: Brian Tyler Date: Fri, 24 Jul 2026 14:16:53 +0100 Subject: [PATCH 2/2] fix: dashboard concurrency group error --- ...gresConcurrencyGroupInspectionOperation.cs | 6 ++-- .../InspectionContractTests.cs | 28 +++++++++++++++++++ 2 files changed, 32 insertions(+), 2 deletions(-) diff --git a/src/Sheddueller.Postgres/Internal/Operations/PostgresConcurrencyGroupInspectionOperation.cs b/src/Sheddueller.Postgres/Internal/Operations/PostgresConcurrencyGroupInspectionOperation.cs index cbd0f2e..9b4dfa8 100644 --- a/src/Sheddueller.Postgres/Internal/Operations/PostgresConcurrencyGroupInspectionOperation.cs +++ b/src/Sheddueller.Postgres/Internal/Operations/PostgresConcurrencyGroupInspectionOperation.cs @@ -256,8 +256,10 @@ group_state as ( concurrency_group.effective_rate_permit_count, concurrency_group.effective_rate_period, concurrency_group.rate_theoretical_arrival_at_utc, - concurrency_group.effective_rate_permit_count is not null - and concurrency_group.rate_theoretical_arrival_at_utc > clock_timestamp() as is_rate_limited, + coalesce( + concurrency_group.effective_rate_permit_count is not null + and concurrency_group.rate_theoretical_arrival_at_utc > clock_timestamp(), + false) as is_rate_limited, concurrency_group.updated_at_utc from group_keys left join {context.Names.ConcurrencyGroups} concurrency_group on concurrency_group.group_key = group_keys.group_key diff --git a/test/Sheddueller.ProviderContracts/InspectionContractTests.cs b/test/Sheddueller.ProviderContracts/InspectionContractTests.cs index 500182a..e6feffa 100644 --- a/test/Sheddueller.ProviderContracts/InspectionContractTests.cs +++ b/test/Sheddueller.ProviderContracts/InspectionContractTests.cs @@ -625,6 +625,34 @@ public async Task ConcurrencyGroupView_DefaultAndOverrideLimits_AreVisible() summary => summary.EffectiveLimit.ShouldBe(5)); } + [Fact] + public async Task ConcurrencyGroupView_UnusedRateLimit_ShowsAvailablePermit() + { + await using var context = await this.CreateContextAsync(); + var rateLimit = new ConcurrencyGroupRateLimit(1, TimeSpan.FromSeconds(1)); + var queued = Guid.NewGuid(); + await context.Store.SetConcurrencyDefaultRateLimitAsync( + new SetConcurrencyDefaultRateLimitRequest("api", rateLimit, DateTimeOffset.UtcNow)); + await context.Store.EnqueueAsync(CreateRequest(queued, groupKeys: ["api"])); + + var page = await context.ConcurrencyGroupReader.SearchConcurrencyGroupsAsync( + new ConcurrencyGroupInspectionQuery( + GroupKey: "api", + IsRateLimited: false, + HasBlockedJobs: false)); + var detail = await context.ConcurrencyGroupReader.GetConcurrencyGroupAsync("api"); + + page.Groups.ShouldHaveSingleItem().GroupKey.ShouldBe("api"); + detail.ShouldNotBeNull(); + detail.Summary.EffectiveRateLimit.ShouldBe(rateLimit); + detail.Summary.NextRatePermitAtUtc.ShouldBeNull(); + detail.Summary.IsRateLimited.ShouldBeFalse(); + detail.Summary.RateBlockedJobCount.ShouldBe(0); + detail.Summary.BlockedJobCount.ShouldBe(0); + detail.RateBlockedJobIds.ShouldBeEmpty(); + detail.BlockedJobIds.ShouldBeEmpty(); + } + [Fact] public async Task ConcurrencyGroupView_RateLimitedGroup_ShowsRateStateAndBlockedJobs() {