From a3c57f9376a4524705968370c13e2ec14fbbe100 Mon Sep 17 00:00:00 2001 From: ducdaiii Date: Fri, 15 Aug 2025 22:23:34 +0700 Subject: [PATCH 01/10] Add HTTP Client Wrapper module: retry, serialization, exception handling --- .../CrispyWaffle.HttpClient.csproj | 13 + .../Exceptions/HttpClientException.cs | 18 ++ .../HttpClientWrapper.cs | 267 ++++++++++++++++++ .../HttpRequestOptions.cs | 44 +++ .../HttpResponse{T}.cs | 46 +++ .../IHttpClientWrapper.cs | 24 ++ Src/CrispyWaffle.HttpClient/RetryPolicy.cs | 129 +++++++++ .../Serialization/IJsonSerializer.cs | 11 + .../Serialization/SystemTextJsonSerializer.cs | 37 +++ .../ServiceCollectionExtensions.cs | 24 ++ 10 files changed, 613 insertions(+) create mode 100644 Src/CrispyWaffle.HttpClient/CrispyWaffle.HttpClient.csproj create mode 100644 Src/CrispyWaffle.HttpClient/Exceptions/HttpClientException.cs create mode 100644 Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs create mode 100644 Src/CrispyWaffle.HttpClient/HttpRequestOptions.cs create mode 100644 Src/CrispyWaffle.HttpClient/HttpResponse{T}.cs create mode 100644 Src/CrispyWaffle.HttpClient/IHttpClientWrapper.cs create mode 100644 Src/CrispyWaffle.HttpClient/RetryPolicy.cs create mode 100644 Src/CrispyWaffle.HttpClient/Serialization/IJsonSerializer.cs create mode 100644 Src/CrispyWaffle.HttpClient/Serialization/SystemTextJsonSerializer.cs create mode 100644 Src/CrispyWaffle.HttpClient/ServiceCollectionExtensions.cs diff --git a/Src/CrispyWaffle.HttpClient/CrispyWaffle.HttpClient.csproj b/Src/CrispyWaffle.HttpClient/CrispyWaffle.HttpClient.csproj new file mode 100644 index 00000000..72c606f6 --- /dev/null +++ b/Src/CrispyWaffle.HttpClient/CrispyWaffle.HttpClient.csproj @@ -0,0 +1,13 @@ + + + + net8.0 + enable + enable + + + + + + + diff --git a/Src/CrispyWaffle.HttpClient/Exceptions/HttpClientException.cs b/Src/CrispyWaffle.HttpClient/Exceptions/HttpClientException.cs new file mode 100644 index 00000000..9fb8a6e7 --- /dev/null +++ b/Src/CrispyWaffle.HttpClient/Exceptions/HttpClientException.cs @@ -0,0 +1,18 @@ +using System; +using System.Net; + +namespace CrispyWaffle.HttpClient.Exceptions +{ + public class HttpClientException : Exception + { + public HttpStatusCode? StatusCode { get; } + public string? ResponseContent { get; } + + public HttpClientException(string message, HttpStatusCode? statusCode = null, string? responseContent = null, Exception? inner = null) + : base(message, inner) + { + StatusCode = statusCode; + ResponseContent = responseContent; + } + } +} diff --git a/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs b/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs new file mode 100644 index 00000000..4b85f6b8 --- /dev/null +++ b/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs @@ -0,0 +1,267 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Net; +using System.Net.Http; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using CrispyWaffle.HttpClient.Exceptions; +using CrispyWaffle.HttpClient.Serialization; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; + +namespace CrispyWaffle.HttpClient +{ + /// + /// Robust HttpClient wrapper with retry/backoff and pluggable serializer. + /// + public class HttpClientWrapper : IHttpClientWrapper + { + private readonly IHttpClientFactory _httpClientFactory; + private readonly HttpRequestOptions _defaults; + private readonly IJsonSerializer _defaultSerializer; + private readonly ILogger? _logger; + + public HttpClientWrapper(IHttpClientFactory httpClientFactory, IOptions options, IJsonSerializer? serializer = null, ILogger? logger = null) + { + _httpClientFactory = httpClientFactory ?? throw new ArgumentNullException(nameof(httpClientFactory)); + _defaults = options?.Value ?? new HttpRequestOptions(); + _defaultSerializer = serializer ?? new SystemTextJsonSerializer(); + _logger = logger; + } + + private System.Net.Http.HttpClient CreateClient(HttpRequestOptions? perRequest) + { + var name = perRequest?.NamedHttpClient ?? _defaults.NamedHttpClient; + System.Net.Http.HttpClient client = string.IsNullOrEmpty(name) + ? _httpClientFactory.CreateClient() + : _httpClientFactory.CreateClient(name); + + var baseAddr = perRequest?.BaseAddress ?? _defaults.BaseAddress; + if (!string.IsNullOrEmpty(baseAddr) && client.BaseAddress == null) + { + client.BaseAddress = new Uri(baseAddr); + } + + var timeoutSeconds = perRequest?.TimeoutSeconds ?? _defaults.TimeoutSeconds; + client.Timeout = TimeSpan.FromSeconds(Math.Max(1, timeoutSeconds)); + + // Merge default headers (global) and per-request headers + var headers = new Dictionary(); + if (_defaults.DefaultRequestHeaders != null) + { + foreach (var kv in _defaults.DefaultRequestHeaders) + headers[kv.Key] = kv.Value; + } + if (perRequest?.DefaultRequestHeaders != null) + { + foreach (var kv in perRequest.DefaultRequestHeaders) + headers[kv.Key] = kv.Value; + } + + // Set headers (be careful to not duplicate) + foreach (var kv in headers) + { + if (!client.DefaultRequestHeaders.Contains(kv.Key)) + { + try + { + client.DefaultRequestHeaders.Add(kv.Key, kv.Value); + } + catch (Exception ex) + { + _logger?.LogWarning(ex, "Failed to set default header {Header}", kv.Key); + } + } + } + + // Ensure Accept header + if (!client.DefaultRequestHeaders.Accept.Any()) + { + client.DefaultRequestHeaders.Accept.Add(new System.Net.Http.Headers.MediaTypeWithQualityHeaderValue("application/json")); + } + + return client; + } + + #region Public Async Methods + + public Task> GetAsync(string url, HttpRequestOptions? options = null, CancellationToken cancellationToken = default) + => SendAsync(HttpMethod.Get, url, null, options, cancellationToken); + + public Task> PostAsync(string url, TRequest? body = default, HttpRequestOptions? options = null, CancellationToken cancellationToken = default) + => SendAsync(HttpMethod.Post, url, body, options, cancellationToken); + + public Task> PutAsync(string url, TRequest? body = default, HttpRequestOptions? options = null, CancellationToken cancellationToken = default) + => SendAsync(HttpMethod.Put, url, body, options, cancellationToken); + + public Task> DeleteAsync(string url, HttpRequestOptions? options = null, CancellationToken cancellationToken = default) + => SendAsync(HttpMethod.Delete, url, null, options, cancellationToken); + + #endregion + + #region Public Sync wrappers (use with caution) + + public HttpResponse Get(string url, HttpRequestOptions? options = null) + => GetAsync(url, options).GetAwaiter().GetResult(); + + public HttpResponse Post(string url, TRequest? body = default, HttpRequestOptions? options = null) + => PostAsync(url, body, options).GetAwaiter().GetResult(); + + public HttpResponse Put(string url, TRequest? body = default, HttpRequestOptions? options = null) + => PutAsync(url, body, options).GetAwaiter().GetResult(); + + public HttpResponse Delete(string url, HttpRequestOptions? options = null) + => DeleteAsync(url, options).GetAwaiter().GetResult(); + + #endregion + + #region Core Send + + private async Task> SendAsync(HttpMethod method, string url, object? body, HttpRequestOptions? perRequest, CancellationToken cancellationToken) + { + var effectiveOptions = MergeOptions(_defaults, perRequest); + var serializer = perRequest?.Serializer ?? _defaultSerializer; + var client = CreateClient(perRequest); + var sw = Stopwatch.StartNew(); + + Func> action = async ct => + { + using var req = new HttpRequestMessage(method, url); + + if (body != null) + { + string payload = serializer.Serialize(body); + req.Content = new StringContent(payload, Encoding.UTF8, effectiveOptions.UseJsonContentType ? "application/json" : "text/plain"); + } + + // Apply per-request headers (non-default) + if (perRequest?.DefaultRequestHeaders != null) + { + foreach (var kv in perRequest.DefaultRequestHeaders) + { + if (!req.Headers.Contains(kv.Key)) + req.Headers.TryAddWithoutValidation(kv.Key, kv.Value); + } + } + + // send + var response = await client.SendAsync(req, HttpCompletionOption.ResponseContentRead, ct).ConfigureAwait(false); + return response; + }; + + var (resp, ex) = await RetryPolicy.ExecuteAsync( + action, + effectiveOptions.RetryCount, + effectiveOptions.RetryBaseDelayMs, + effectiveOptions.MaxRetryDelayMs, + cancellationToken).ConfigureAwait(false); + + sw.Stop(); + + if (resp != null) + { + var raw = resp.Content != null ? await resp.Content.ReadAsStringAsync().ConfigureAwait(false) : null; + var headers = resp.Headers.ToDictionary(h => h.Key, h => h.Value.AsEnumerable()); + // include content headers too + if (resp.Content?.Headers != null) + { + foreach (var h in resp.Content.Headers) + headers[h.Key] = h.Value.AsEnumerable(); + } + + if (resp.IsSuccessStatusCode) + { + // Try deserialize; if empty raw -> default(T) + T? data = default; + try + { + if (!string.IsNullOrWhiteSpace(raw)) + { + data = serializer.Deserialize(raw); + } + } + catch (Exception deserEx) + { + _logger?.LogError(deserEx, "Failed to deserialize response body to {Type}", typeof(T).FullName); + // We treat deserialization failure as an error + if (effectiveOptions.ThrowOnError) + { + throw new HttpClientException($"Failed to deserialize response to {typeof(T)}", resp.StatusCode, raw, deserEx); + } + + return HttpResponse.Failure(resp.StatusCode, raw, new List { "DeserializationFailed: " + deserEx.Message }, headers, sw.Elapsed); + } + + return HttpResponse.Success(data, resp.StatusCode, raw, headers, sw.Elapsed); + } + else + { + var errors = new List { $"HTTP {(int)resp.StatusCode} {resp.ReasonPhrase}" }; + if (!string.IsNullOrWhiteSpace(raw)) + errors.Add(raw); + + if (effectiveOptions.ThrowOnError) + { + throw new HttpClientException($"HTTP request failed with {(int)resp.StatusCode}", resp.StatusCode, raw); + } + + return HttpResponse.Failure(resp.StatusCode, raw, errors, headers, sw.Elapsed); + } + } + else + { + // No response (exception) + if (effectiveOptions.ThrowOnError) + { + throw new HttpClientException("HTTP request failed after retries", null, null, ex); + } + + var errors = new List { ex?.Message ?? "Unknown error" }; + return HttpResponse.Failure(null, null, errors, null, sw.Elapsed); + } + } + + #endregion + + #region Helpers + + private static HttpRequestOptions MergeOptions(HttpRequestOptions defaults, HttpRequestOptions? perRequest) + { + if (perRequest == null) return defaults; + // create new instance merging properties - only simple properties merged here + return new HttpRequestOptions + { + BaseAddress = perRequest.BaseAddress ?? defaults.BaseAddress, + TimeoutSeconds = perRequest.TimeoutSeconds != default ? perRequest.TimeoutSeconds : defaults.TimeoutSeconds, + RetryCount = perRequest.RetryCount != default ? perRequest.RetryCount : defaults.RetryCount, + RetryBaseDelayMs = perRequest.RetryBaseDelayMs != default ? perRequest.RetryBaseDelayMs : defaults.RetryBaseDelayMs, + MaxRetryDelayMs = perRequest.MaxRetryDelayMs != default ? perRequest.MaxRetryDelayMs : defaults.MaxRetryDelayMs, + ThrowOnError = perRequest.ThrowOnError, + DefaultRequestHeaders = MergeHeaders(defaults.DefaultRequestHeaders, perRequest.DefaultRequestHeaders), + NamedHttpClient = perRequest.NamedHttpClient ?? defaults.NamedHttpClient, + UseJsonContentType = perRequest.UseJsonContentType, + Serializer = perRequest.Serializer ?? defaults.Serializer + }; + } + + private static IDictionary? MergeHeaders(IDictionary? a, IDictionary? b) + { + if (a == null && b == null) return null; + var dict = new Dictionary(StringComparer.OrdinalIgnoreCase); + if (a != null) + { + foreach (var kv in a) dict[kv.Key] = kv.Value; + } + if (b != null) + { + foreach (var kv in b) dict[kv.Key] = kv.Value; + } + return dict; + } + + #endregion + } +} diff --git a/Src/CrispyWaffle.HttpClient/HttpRequestOptions.cs b/Src/CrispyWaffle.HttpClient/HttpRequestOptions.cs new file mode 100644 index 00000000..0f497e39 --- /dev/null +++ b/Src/CrispyWaffle.HttpClient/HttpRequestOptions.cs @@ -0,0 +1,44 @@ +using System; +using System.Collections.Generic; +using System.Net.Http.Headers; +using System.Text.Json; +using CrispyWaffle.HttpClient.Serialization; + +namespace CrispyWaffle.HttpClient +{ + /// + /// Options used as defaults (via IOptions) and can be overriden per-request. + /// + public class HttpRequestOptions + { + /// Gets or sets base address for created HttpClient if specified (can be null when using full URLs). + public string? BaseAddress { get; set; } + + /// Gets or sets default timeout in seconds. + public int TimeoutSeconds { get; set; } = 30; + + /// Gets or sets number of retry attempts on transient failures (0 = no retry). + public int RetryCount { get; set; } = 3; + + /// Gets or sets base delay in milliseconds used for exponential backoff. + public int RetryBaseDelayMs { get; set; } = 200; // 200ms base + + /// Gets or sets maximum allowed delay between retries in ms. + public int MaxRetryDelayMs { get; set; } = 10000; // 10s + + /// Gets or sets a value indicating whether if true, methods throw HttpClientException on final failure instead of returning unsuccessful HttpResponse<T>. + public bool ThrowOnError { get; set; } = false; + + /// Gets or sets default request headers to add to each request if present. + public IDictionary? DefaultRequestHeaders { get; set; } + + /// Gets or sets optional specific named client to create from IHttpClientFactory. If null, factory.CreateClient() is used. + public string? NamedHttpClient { get; set; } + + /// Gets or sets a value indicating whether if true, sends content type application/json when body present. + public bool UseJsonContentType { get; set; } = true; + + /// Gets or sets optional serializer override per-request. If null will use DI resolved one. + public IJsonSerializer? Serializer { get; set; } + } +} \ No newline at end of file diff --git a/Src/CrispyWaffle.HttpClient/HttpResponse{T}.cs b/Src/CrispyWaffle.HttpClient/HttpResponse{T}.cs new file mode 100644 index 00000000..75570cf9 --- /dev/null +++ b/Src/CrispyWaffle.HttpClient/HttpResponse{T}.cs @@ -0,0 +1,46 @@ +using System; +using System.Collections.Generic; +using System.Net; + +namespace CrispyWaffle.HttpClient +{ + /// + /// Generic response wrapper returned by the wrapper methods. + /// + public class HttpResponse + { + public bool IsSuccess { get; set; } + public HttpStatusCode? StatusCode { get; set; } + public T? Data { get; set; } + public string? RawContent { get; set; } + public IList? Errors { get; set; } + public IDictionary>? Headers { get; set; } + public TimeSpan Duration { get; set; } + + public static HttpResponse Success(T? data, HttpStatusCode statusCode, string? rawContent, IDictionary>? headers, TimeSpan duration) + { + return new HttpResponse + { + IsSuccess = true, + StatusCode = statusCode, + Data = data, + RawContent = rawContent, + Headers = headers, + Duration = duration + }; + } + + public static HttpResponse Failure(HttpStatusCode? statusCode, string? rawContent, IList? errors, IDictionary>? headers, TimeSpan duration) + { + return new HttpResponse + { + IsSuccess = false, + StatusCode = statusCode, + RawContent = rawContent, + Errors = errors, + Headers = headers, + Duration = duration + }; + } + } +} \ No newline at end of file diff --git a/Src/CrispyWaffle.HttpClient/IHttpClientWrapper.cs b/Src/CrispyWaffle.HttpClient/IHttpClientWrapper.cs new file mode 100644 index 00000000..8a0a3748 --- /dev/null +++ b/Src/CrispyWaffle.HttpClient/IHttpClientWrapper.cs @@ -0,0 +1,24 @@ +using System; +using System.Net.Http; +using System.Threading; +using System.Threading.Tasks; + +namespace CrispyWaffle.HttpClient +{ + /// + /// High level HTTP client wrapper - primarily async methods (sync wrappers provided but not recommended). + /// + public interface IHttpClientWrapper + { + Task> GetAsync(string url, HttpRequestOptions? options = null, CancellationToken cancellationToken = default); + Task> PostAsync(string url, TRequest? body = default, HttpRequestOptions? options = null, CancellationToken cancellationToken = default); + Task> PutAsync(string url, TRequest? body = default, HttpRequestOptions? options = null, CancellationToken cancellationToken = default); + Task> DeleteAsync(string url, HttpRequestOptions? options = null, CancellationToken cancellationToken = default); + + // sync helpers (block on async) - use with caution (may deadlock in sync contexts) + HttpResponse Get(string url, HttpRequestOptions? options = null); + HttpResponse Post(string url, TRequest? body = default, HttpRequestOptions? options = null); + HttpResponse Put(string url, TRequest? body = default, HttpRequestOptions? options = null); + HttpResponse Delete(string url, HttpRequestOptions? options = null); + } +} \ No newline at end of file diff --git a/Src/CrispyWaffle.HttpClient/RetryPolicy.cs b/Src/CrispyWaffle.HttpClient/RetryPolicy.cs new file mode 100644 index 00000000..2186c85e --- /dev/null +++ b/Src/CrispyWaffle.HttpClient/RetryPolicy.cs @@ -0,0 +1,129 @@ +using System; +using System.Net; +using System.Net.Http; +using System.Threading; +using System.Threading.Tasks; + +namespace CrispyWaffle.HttpClient +{ + /// + /// Small retry helper with exponential backoff + jitter and handling Retry-After header for 429. + /// No external dependency required. + /// + internal static class RetryPolicy + { + public static async Task<(HttpResponseMessage? Response, Exception? Exception)> ExecuteAsync( + Func> action, + int retryCount, + int baseDelayMs, + int maxDelayMs, + CancellationToken ct) + { + if (retryCount <= 0) + { + try + { + var r = await action(ct).ConfigureAwait(false); + return (r, null); + } + catch (Exception ex) + { + return (null, ex); + } + } + + var rnd = new Random(); + Exception? lastEx = null; + + for (int attempt = 0; attempt <= retryCount; attempt++) + { + try + { + var response = await action(ct).ConfigureAwait(false); + + if (!ShouldRetryResponse(response)) + { + return (response, null); + } + + // Will retry + var delay = ComputeDelay(attempt, baseDelayMs, maxDelayMs, rnd); + + // If 429 and Retry-After header present, try to honor it + if (response.StatusCode == (HttpStatusCode)429) + { + var ra = GetRetryAfterDelay(response); + if (ra.HasValue) delay = Math.Min((int)ra.Value.TotalMilliseconds, maxDelayMs); + } + + await Task.Delay(delay, ct).ConfigureAwait(false); + continue; + } + catch (Exception ex) when (IsTransientException(ex)) + { + lastEx = ex; + if (attempt == retryCount) break; + + var delay = ComputeDelay(attempt, baseDelayMs, maxDelayMs, rnd); + await Task.Delay(delay, ct).ConfigureAwait(false); + continue; + } + } + + return (null, lastEx); + } + + private static bool ShouldRetryResponse(HttpResponseMessage response) + { + if (response == null) return true; + var code = (int)response.StatusCode; + // Retry for 5xx, 408 Request Timeout, 429 Too Many Requests + if (code >= 500 || code == 408 || code == 429) return true; + return false; + } + + private static bool IsTransientException(Exception ex) + { + // HttpRequestException and TaskCanceled (timeout) are considered transient + return ex is HttpRequestException || ex is OperationCanceledException || ex is TimeoutException; + } + + private static int ComputeDelay(int attempt, int baseDelayMs, int maxDelayMs, Random rnd) + { + // exponential backoff with jitter + // formula: min(maxDelay, base * 2^attempt) +/- jitter up to 20% + var exponential = baseDelayMs * (1 << Math.Min(attempt, 10)); // cap shift + var capped = Math.Min(exponential, maxDelayMs); + var jitter = (int)(capped * 0.2 * (rnd.NextDouble() - 0.5)); // ±10% + var result = Math.Max(0, capped + jitter); + return result; + } + + private static TimeSpan? GetRetryAfterDelay(HttpResponseMessage response) + { + // 1. Chuẩn HTTP (HttpResponseHeaders) + if (response.Headers?.RetryAfter != null) + { + var ra = response.Headers.RetryAfter; + if (ra.Delta.HasValue) + return ra.Delta.Value; + if (ra.Date.HasValue) + return ra.Date.Value - DateTimeOffset.UtcNow; + } + + // 2. Fallback - parse thủ công từ mọi header (cả ContentHeaders) + if (response.Headers.TryGetValues("Retry-After", out var headerValues) || + (response.Content?.Headers?.TryGetValues("Retry-After", out headerValues) ?? false)) + { + var retryValue = headerValues.FirstOrDefault(); + if (int.TryParse(retryValue, out var seconds)) + return TimeSpan.FromSeconds(seconds); + + if (DateTimeOffset.TryParse(retryValue, out var retryDate)) + return retryDate - DateTimeOffset.UtcNow; + } + + return null; + } + } +} \ No newline at end of file diff --git a/Src/CrispyWaffle.HttpClient/Serialization/IJsonSerializer.cs b/Src/CrispyWaffle.HttpClient/Serialization/IJsonSerializer.cs new file mode 100644 index 00000000..7cc46edf --- /dev/null +++ b/Src/CrispyWaffle.HttpClient/Serialization/IJsonSerializer.cs @@ -0,0 +1,11 @@ +using System; + +namespace CrispyWaffle.HttpClient.Serialization +{ + public interface IJsonSerializer + { + string Serialize(T value); + T? Deserialize(string json); + object? Deserialize(string json, Type type); + } +} \ No newline at end of file diff --git a/Src/CrispyWaffle.HttpClient/Serialization/SystemTextJsonSerializer.cs b/Src/CrispyWaffle.HttpClient/Serialization/SystemTextJsonSerializer.cs new file mode 100644 index 00000000..c3c76e91 --- /dev/null +++ b/Src/CrispyWaffle.HttpClient/Serialization/SystemTextJsonSerializer.cs @@ -0,0 +1,37 @@ +using System; +using System.Text.Json; + +namespace CrispyWaffle.HttpClient.Serialization +{ + public class SystemTextJsonSerializer : IJsonSerializer + { + private readonly JsonSerializerOptions _options; + + public SystemTextJsonSerializer(JsonSerializerOptions? options = null) + { + _options = options ?? new JsonSerializerOptions + { + PropertyNameCaseInsensitive = true, + PropertyNamingPolicy = JsonNamingPolicy.CamelCase, + DefaultIgnoreCondition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingNull + }; + } + + public string Serialize(T value) + { + return JsonSerializer.Serialize(value, _options); + } + + public T? Deserialize(string json) + { + if (string.IsNullOrWhiteSpace(json)) return default; + return JsonSerializer.Deserialize(json, _options); + } + + public object? Deserialize(string json, Type type) + { + if (string.IsNullOrWhiteSpace(json)) return null; + return JsonSerializer.Deserialize(json, type, _options); + } + } +} \ No newline at end of file diff --git a/Src/CrispyWaffle.HttpClient/ServiceCollectionExtensions.cs b/Src/CrispyWaffle.HttpClient/ServiceCollectionExtensions.cs new file mode 100644 index 00000000..b7ac66c2 --- /dev/null +++ b/Src/CrispyWaffle.HttpClient/ServiceCollectionExtensions.cs @@ -0,0 +1,24 @@ +using System; +using Microsoft.Extensions.DependencyInjection; + +namespace CrispyWaffle.HttpClient +{ + public static class ServiceCollectionExtensions + { + /// + /// Registers HttpClient wrapper with defaults and adds a system-text JSON serializer. + /// Ensure you called services.AddHttpClient() earlier / or it will be added automatically. + /// + public static IServiceCollection AddCrispyWaffleHttpClient(this IServiceCollection services, Action? configureOptions = null) + { + services.AddHttpClient(); // safe to call multiple times + var options = new HttpRequestOptions(); + configureOptions?.Invoke(options); + services.Configure(configureOptions ?? (_ => { })); + // Serializer registration + services.AddSingleton(); + services.AddSingleton(); + return services; + } + } +} \ No newline at end of file From ea773ecd3b907769c30a64787a3d0ba1d711c0a4 Mon Sep 17 00:00:00 2001 From: ducdaiii Date: Fri, 15 Aug 2025 22:28:28 +0700 Subject: [PATCH 02/10] Add HTTP Client Wrapper module: retry, serialization, exception handling --- Directory.Packages.props | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/Directory.Packages.props b/Directory.Packages.props index c70560bc..9dd7efa9 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -14,12 +14,14 @@ + + - + @@ -33,4 +35,4 @@ runtime; build; native; contentfiles; analyzers; buildtransitive - + \ No newline at end of file From 26a411d60e801aa4f100eea75a4ef525a66d70d7 Mon Sep 17 00:00:00 2001 From: codefactor-io Date: Fri, 15 Aug 2025 15:32:14 +0000 Subject: [PATCH 03/10] [CodeFactor] Apply fixes --- Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs b/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs index 4b85f6b8..8e48bdd5 100644 --- a/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs +++ b/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Collections.Generic; using System.Diagnostics; using System.Linq; @@ -78,7 +78,7 @@ private System.Net.Http.HttpClient CreateClient(HttpRequestOptions? perRequest) } // Ensure Accept header - if (!client.DefaultRequestHeaders.Accept.Any()) + if (client.DefaultRequestHeaders.Accept.Count == 0) { client.DefaultRequestHeaders.Accept.Add(new System.Net.Http.Headers.MediaTypeWithQualityHeaderValue("application/json")); } From 7219c84a23709a9a78c8bd1e42f846f3da6251d2 Mon Sep 17 00:00:00 2001 From: ducdaiii Date: Fri, 15 Aug 2025 22:50:26 +0700 Subject: [PATCH 04/10] Refactor: HTTP Client Wrapper module: retry, serialization, exception handling --- .../HttpClientWrapper.cs | 87 +++++++++++++------ Src/CrispyWaffle.HttpClient/RetryPolicy.cs | 2 +- .../ServiceCollectionExtensions.cs | 39 +++++++-- 3 files changed, 93 insertions(+), 35 deletions(-) diff --git a/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs b/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs index 8e48bdd5..d54d404a 100644 --- a/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs +++ b/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs @@ -122,36 +122,46 @@ public HttpResponse Put(string url, TRequest? bo private async Task> SendAsync(HttpMethod method, string url, object? body, HttpRequestOptions? perRequest, CancellationToken cancellationToken) { + // Merge defaults and per-request options first. var effectiveOptions = MergeOptions(_defaults, perRequest); - var serializer = perRequest?.Serializer ?? _defaultSerializer; - var client = CreateClient(perRequest); + + // Use serializer from merged options, fallback to default. + var serializer = effectiveOptions.Serializer ?? _defaultSerializer; + + // Create HttpClient using the effective options (so named client, base address, timeout, etc. apply). + var client = CreateClient(effectiveOptions); + var sw = Stopwatch.StartNew(); + // The action will create a NEW HttpRequestMessage each time it's invoked (important for retries). Func> action = async ct => { + // Create a new request per attempt - ensures headers/content are fresh for retries. using var req = new HttpRequestMessage(method, url); if (body != null) { + // Serialize the body (synchronous serializer expected here). string payload = serializer.Serialize(body); req.Content = new StringContent(payload, Encoding.UTF8, effectiveOptions.UseJsonContentType ? "application/json" : "text/plain"); } - // Apply per-request headers (non-default) - if (perRequest?.DefaultRequestHeaders != null) + // Apply per-request (merged) headers to the request message. + if (effectiveOptions.DefaultRequestHeaders != null) { - foreach (var kv in perRequest.DefaultRequestHeaders) + foreach (var kv in effectiveOptions.DefaultRequestHeaders) { if (!req.Headers.Contains(kv.Key)) req.Headers.TryAddWithoutValidation(kv.Key, kv.Value); } } - // send - var response = await client.SendAsync(req, HttpCompletionOption.ResponseContentRead, ct).ConfigureAwait(false); - return response; + // Send with ResponseHeadersRead to start streaming the body (we will read it explicitly afterward). + // This avoids buffering the entire response in some handlers. + return await client.SendAsync(req, HttpCompletionOption.ResponseHeadersRead, ct).ConfigureAwait(false); }; + // Execute the action with retry policy. The policy is expected to return the final response (or null + exception). var (resp, ex) = await RetryPolicy.ExecuteAsync( action, effectiveOptions.RetryCount, @@ -161,23 +171,58 @@ private async Task> SendAsync(HttpMethod method, string url, sw.Stop(); - if (resp != null) + // If no response after retries + if (resp == null) { - var raw = resp.Content != null ? await resp.Content.ReadAsStringAsync().ConfigureAwait(false) : null; + if (effectiveOptions.ThrowOnError) + { + throw new HttpClientException("HTTP request failed after retries", null, null, ex); + } + + var errors = new List { ex?.Message ?? "Unknown error" }; + return HttpResponse.Failure(null, null, errors, null, sw.Elapsed); + } + + // IMPORTANT: dispose HttpResponseMessage after we've finished reading headers/body. + using (resp) + { + string? raw = null; + try + { + // If there is content, read it as string. Keep it simple and synchronous from serializer's perspective. + if (resp.Content != null) + { + raw = await resp.Content.ReadAsStringAsync().ConfigureAwait(false); + } + } + catch (Exception readEx) + { + // Failed to read content. + _logger?.LogError(readEx, "Failed to read response content for {Method} {Url}", method, url); + if (effectiveOptions.ThrowOnError) + { + throw new HttpClientException("Failed to read response content", resp.StatusCode, null, readEx); + } + + var readErrors = new List { readEx.Message }; + return HttpResponse.Failure(resp.StatusCode, null, readErrors, resp.Headers.ToDictionary(h => h.Key, h => h.Value.AsEnumerable()), sw.Elapsed); + } + + // Gather headers (including content headers). var headers = resp.Headers.ToDictionary(h => h.Key, h => h.Value.AsEnumerable()); - // include content headers too if (resp.Content?.Headers != null) { foreach (var h in resp.Content.Headers) headers[h.Key] = h.Value.AsEnumerable(); } + // Successful HTTP status if (resp.IsSuccessStatusCode) { - // Try deserialize; if empty raw -> default(T) T? data = default; try { + // If raw body is present, try to deserialize; otherwise default(T). if (!string.IsNullOrWhiteSpace(raw)) { data = serializer.Deserialize(raw); @@ -185,8 +230,8 @@ private async Task> SendAsync(HttpMethod method, string url, } catch (Exception deserEx) { - _logger?.LogError(deserEx, "Failed to deserialize response body to {Type}", typeof(T).FullName); - // We treat deserialization failure as an error + // Deserialization failed - log and either throw or return failure depending on options. + _logger?.LogError(deserEx, "Failed to deserialize response body to {Type} for {Method} {Url}", typeof(T).FullName, method, url); if (effectiveOptions.ThrowOnError) { throw new HttpClientException($"Failed to deserialize response to {typeof(T)}", resp.StatusCode, raw, deserEx); @@ -199,6 +244,7 @@ private async Task> SendAsync(HttpMethod method, string url, } else { + // Non-success HTTP status var errors = new List { $"HTTP {(int)resp.StatusCode} {resp.ReasonPhrase}" }; if (!string.IsNullOrWhiteSpace(raw)) errors.Add(raw); @@ -211,17 +257,6 @@ private async Task> SendAsync(HttpMethod method, string url, return HttpResponse.Failure(resp.StatusCode, raw, errors, headers, sw.Elapsed); } } - else - { - // No response (exception) - if (effectiveOptions.ThrowOnError) - { - throw new HttpClientException("HTTP request failed after retries", null, null, ex); - } - - var errors = new List { ex?.Message ?? "Unknown error" }; - return HttpResponse.Failure(null, null, errors, null, sw.Elapsed); - } } #endregion @@ -264,4 +299,4 @@ private static HttpRequestOptions MergeOptions(HttpRequestOptions defaults, Http #endregion } -} +} \ No newline at end of file diff --git a/Src/CrispyWaffle.HttpClient/RetryPolicy.cs b/Src/CrispyWaffle.HttpClient/RetryPolicy.cs index 2186c85e..4d44f4ff 100644 --- a/Src/CrispyWaffle.HttpClient/RetryPolicy.cs +++ b/Src/CrispyWaffle.HttpClient/RetryPolicy.cs @@ -32,7 +32,7 @@ internal static class RetryPolicy } } - var rnd = new Random(); + var rnd = Random.Shared; Exception? lastEx = null; for (int attempt = 0; attempt <= retryCount; attempt++) diff --git a/Src/CrispyWaffle.HttpClient/ServiceCollectionExtensions.cs b/Src/CrispyWaffle.HttpClient/ServiceCollectionExtensions.cs index b7ac66c2..0c3e1cfa 100644 --- a/Src/CrispyWaffle.HttpClient/ServiceCollectionExtensions.cs +++ b/Src/CrispyWaffle.HttpClient/ServiceCollectionExtensions.cs @@ -6,18 +6,41 @@ namespace CrispyWaffle.HttpClient public static class ServiceCollectionExtensions { /// - /// Registers HttpClient wrapper with defaults and adds a system-text JSON serializer. - /// Ensure you called services.AddHttpClient() earlier / or it will be added automatically. + /// Registers the HttpClient wrapper with defaults and a System.Text.Json serializer. + /// Make sure you call this after configuring your application's dependency injection. /// - public static IServiceCollection AddCrispyWaffleHttpClient(this IServiceCollection services, Action? configureOptions = null) + /// + /// Improvements: + /// 1. Use services.Configure(...) to register options in a standard and thread-safe way. + /// 2. Ensure serializer is thread-safe by registering a single instance. + /// 3. Prepare for retry logic improvements in HttpClientWrapper by ensuring that per-request serializers are + /// handled without mutating shared state. + /// + public static IServiceCollection AddCrispyWaffleHttpClient( + this IServiceCollection services, + Action? configureOptions = null) { - services.AddHttpClient(); // safe to call multiple times - var options = new HttpRequestOptions(); - configureOptions?.Invoke(options); - services.Configure(configureOptions ?? (_ => { })); - // Serializer registration + // Ensure HttpClientFactory is registered (safe to call multiple times). + services.AddHttpClient(); + + // Register options using the built-in Microsoft.Extensions.Options pattern. + // This ensures thread-safety and that options can be injected anywhere via IOptions. + if (configureOptions != null) + { + services.Configure(configureOptions); + } + else + { + services.Configure(_ => { }); + } + + // Register serializer as a singleton for thread-safety. + // This assumes the serializer implementation is stateless or internally thread-safe. services.AddSingleton(); + + // Register the HttpClient wrapper. services.AddSingleton(); + return services; } } From ebdc245e8decacbd25f416204adaf8d0c6386d8a Mon Sep 17 00:00:00 2001 From: ducdaiii Date: Fri, 15 Aug 2025 22:53:03 +0700 Subject: [PATCH 05/10] Refactor: HTTP Client Wrapper module: retry, serialization, exception handling --- CrispyWaffle.sln | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/CrispyWaffle.sln b/CrispyWaffle.sln index 8802577c..b49125c6 100644 --- a/CrispyWaffle.sln +++ b/CrispyWaffle.sln @@ -45,6 +45,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "CrispyWaffle.CouchDB", "Src EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "CrispyWaffle.IntegrationTests", "Tests\CrispyWaffle.IntegrationTests\CrispyWaffle.IntegrationTests.csproj", "{0E3B9F7C-D91C-4862-BE7C-58402C3786B3}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "CrispyWaffle.HttpClient", "Src\CrispyWaffle.HttpClient\CrispyWaffle.HttpClient.csproj", "{BBF0D93B-980B-F085-B8D8-247B55FBFD0F}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -107,6 +109,10 @@ Global {0E3B9F7C-D91C-4862-BE7C-58402C3786B3}.Debug|Any CPU.Build.0 = Debug|Any CPU {0E3B9F7C-D91C-4862-BE7C-58402C3786B3}.Release|Any CPU.ActiveCfg = Release|Any CPU {0E3B9F7C-D91C-4862-BE7C-58402C3786B3}.Release|Any CPU.Build.0 = Release|Any CPU + {BBF0D93B-980B-F085-B8D8-247B55FBFD0F}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {BBF0D93B-980B-F085-B8D8-247B55FBFD0F}.Debug|Any CPU.Build.0 = Debug|Any CPU + {BBF0D93B-980B-F085-B8D8-247B55FBFD0F}.Release|Any CPU.ActiveCfg = Release|Any CPU + {BBF0D93B-980B-F085-B8D8-247B55FBFD0F}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -128,6 +134,7 @@ Global {DAE04EC0-911F-11D3-BF4B-00C04F79EFBC} = {2FAD8261-E9AF-4B92-9EBE-35079F6250E8} {7DA4B7D8-6621-43FA-969C-2F919D9ACD4C} = {8121D270-758B-469E-A72C-04E58A06A5EF} {0E3B9F7C-D91C-4862-BE7C-58402C3786B3} = {7D691CCA-0EAF-46F0-8375-5295F2C0FBC2} + {BBF0D93B-980B-F085-B8D8-247B55FBFD0F} = {8121D270-758B-469E-A72C-04E58A06A5EF} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {4C5D9BB9-952C-49AF-A90A-835EAE8680FC} From 748644ef0ec433da22a2f4e4404a114888e85bbe Mon Sep 17 00:00:00 2001 From: ducdaiii Date: Sat, 16 Aug 2025 14:02:03 +0700 Subject: [PATCH 06/10] feat(background-jobs): add full background job module with queue, workers, and persistence --- Directory.Packages.props | 9 +- .../Abstractions/IBackgroundJob.cs | 13 + .../Abstractions/IBackgroundJobHandler.cs | 13 + .../Abstractions/IJobScheduler.cs | 9 + .../Abstractions/IJobStore.cs | 19 ++ .../Abstractions/JobEntity.cs | 47 +++ .../Abstractions/JobPriority.cs | 9 + .../Abstractions/JobStatus.cs | 11 + .../Core/BackgroundJobQueue.cs | 47 +++ .../Core/BackgroundWorker.cs | 278 ++++++++++++++++++ .../Core/IJobHandlerRegistry.cs | 10 + .../Core/JobDispatcher.cs | 77 +++++ .../Core/JobHandlerRegistry.cs | 29 ++ .../Core/JobOptions.cs | 26 ++ .../Core/JobResult.cs | 8 + .../CrispyWaffle.BackgroundJobs.csproj | 20 ++ .../Exceptions/JobFailedException.cs | 7 + .../Extensions/ServiceCollectionExtensions.cs | 90 ++++++ .../Monitoring/JobMetrics.cs | 19 ++ .../Monitoring/MonitoringMiddleware.cs | 29 ++ .../Persistence/EfCoreJobStore.cs | 86 ++++++ .../Persistence/InMemoryJobStore.cs | 74 +++++ .../Persistence/JobDbContext.cs | 31 ++ .../Scheduling/CronJobScheduler.cs | 40 +++ .../Scheduling/DelayedJobScheduler.cs | 31 ++ 25 files changed, 1031 insertions(+), 1 deletion(-) create mode 100644 Src/CrispyWaffle.BackgroundJobs/Abstractions/IBackgroundJob.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Abstractions/IBackgroundJobHandler.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobScheduler.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobStore.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Abstractions/JobEntity.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Abstractions/JobPriority.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Abstractions/JobStatus.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Core/BackgroundJobQueue.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Core/IJobHandlerRegistry.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Core/JobDispatcher.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Core/JobHandlerRegistry.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Core/JobOptions.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Core/JobResult.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/CrispyWaffle.BackgroundJobs.csproj create mode 100644 Src/CrispyWaffle.BackgroundJobs/Exceptions/JobFailedException.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Extensions/ServiceCollectionExtensions.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Monitoring/JobMetrics.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Monitoring/MonitoringMiddleware.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Persistence/EfCoreJobStore.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Persistence/InMemoryJobStore.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Persistence/JobDbContext.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Scheduling/CronJobScheduler.cs create mode 100644 Src/CrispyWaffle.BackgroundJobs/Scheduling/DelayedJobScheduler.cs diff --git a/Directory.Packages.props b/Directory.Packages.props index 9dd7efa9..e984c478 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -14,8 +14,15 @@ + + + + + + - + + diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/IBackgroundJob.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IBackgroundJob.cs new file mode 100644 index 00000000..cc4562a9 --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IBackgroundJob.cs @@ -0,0 +1,13 @@ +using CrispyWaffle.BackgroundJobs.Core; + +namespace CrispyWaffle.BackgroundJobs.Abstractions +{ + /// + /// Low-level job representation (mostly used for in-memory workflows). + /// For persisted jobs, prefer IBackgroundJobHandler with registry-based activation. + /// + public interface IBackgroundJob + { + Task ExecuteAsync(CancellationToken cancellationToken); + } +} \ No newline at end of file diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/IBackgroundJobHandler.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IBackgroundJobHandler.cs new file mode 100644 index 00000000..fe255907 --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IBackgroundJobHandler.cs @@ -0,0 +1,13 @@ +using CrispyWaffle.BackgroundJobs.Core; + +namespace CrispyWaffle.BackgroundJobs.Abstractions +{ + /// + /// Typed handler used by persisted jobs. Implement this to allow DI-resolved handlers. + /// TData is the payload type stored as JSON in the job store. + /// + public interface IBackgroundJobHandler + { + Task HandleAsync(TData data, CancellationToken cancellationToken); + } +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobScheduler.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobScheduler.cs new file mode 100644 index 00000000..f01a9d4a --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobScheduler.cs @@ -0,0 +1,9 @@ +namespace CrispyWaffle.BackgroundJobs.Abstractions +{ + public interface IJobScheduler + { + Task ScheduleAsync(string handlerName, object payload, TimeSpan delay, int maxAttempts = 3, JobPriority priority = JobPriority.Normal); + + Task EnqueueAsync(string handlerName, object payload, int maxAttempts = 3, JobPriority priority = JobPriority.Normal); + } +} \ No newline at end of file diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobStore.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobStore.cs new file mode 100644 index 00000000..7ba99e15 --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobStore.cs @@ -0,0 +1,19 @@ +namespace CrispyWaffle.BackgroundJobs.Abstractions +{ + public interface IJobStore + { + Task SaveAsync(JobEntity job, CancellationToken cancellationToken = default); + + /// + /// Fetch the next available job that is due (ScheduledAt <= UtcNow) and Pending. + /// Implementation should atomically mark it as Processing to avoid double pick-up. + /// + Task FetchNextAsync(CancellationToken cancellationToken = default); + + Task MarkCompletedAsync(Guid jobId, CancellationToken cancellationToken = default); + + Task MarkFailedAsync(Guid jobId, string error, CancellationToken cancellationToken = default); + + Task MarkRetryAsync(Guid jobId, DateTimeOffset? nextAttemptAt, int attemptCount, CancellationToken cancellationToken = default); + } +} \ No newline at end of file diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobEntity.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobEntity.cs new file mode 100644 index 00000000..bf97626f --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobEntity.cs @@ -0,0 +1,47 @@ +namespace CrispyWaffle.BackgroundJobs.Abstractions +{ + using System; + + /// + /// Persistent representation of a job. + /// HandlerName identifies a registered handler; Payload is JSON. + /// + public class JobEntity + { + public Guid Id { get; set; } = Guid.NewGuid(); + + /// + /// Logical handler name registered in IJobHandlerRegistry. + /// + public string HandlerName { get; set; } = string.Empty; + + /// + /// JSON payload (serialized) for the handler. + /// + public string Payload { get; set; } = string.Empty; + + public JobPriority Priority { get; set; } = JobPriority.Normal; + + public JobStatus Status { get; set; } = JobStatus.Pending; + + /// + /// When the job becomes due for execution. Null means immediately. + /// + public DateTimeOffset? ScheduledAt { get; set; } + + public int Attempt { get; set; } = 0; + + public int MaxAttempt { get; set; } = 3; + + public string? LastError { get; set; } + + public DateTimeOffset CreatedAt { get; set; } = DateTimeOffset.UtcNow; + + public DateTimeOffset UpdatedAt { get; set; } = DateTimeOffset.UtcNow; + + /// + /// Optional delay in seconds between retries (used only as hint). + /// + public int RetryDelaySeconds { get; set; } = 0; + } +} \ No newline at end of file diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobPriority.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobPriority.cs new file mode 100644 index 00000000..621587ae --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobPriority.cs @@ -0,0 +1,9 @@ +namespace CrispyWaffle.BackgroundJobs.Abstractions +{ + public enum JobPriority + { + High = 0, + Normal = 1, + Low = 2 + } +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobStatus.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobStatus.cs new file mode 100644 index 00000000..c7ef22d9 --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobStatus.cs @@ -0,0 +1,11 @@ +namespace CrispyWaffle.BackgroundJobs.Abstractions +{ + public enum JobStatus + { + Pending = 0, + Processing = 1, + Completed = 2, + Failed = 3, + Dead = 4 + } +} \ No newline at end of file diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundJobQueue.cs b/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundJobQueue.cs new file mode 100644 index 00000000..a1d01af2 --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundJobQueue.cs @@ -0,0 +1,47 @@ +using CrispyWaffle.BackgroundJobs.Abstractions; +using System.Threading.Channels; + +namespace CrispyWaffle.BackgroundJobs.Core +{ + /// + /// Lightweight in-memory priority queue using Channel for signaling. + /// Priorities: lower int = higher priority. + /// This queue stores JobEntity (for uniformity with persistent store model). + /// + public class BackgroundJobQueue + { + private readonly object _lock = new(); + private readonly PriorityQueue _pq = new(); + private readonly Channel _signal = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = false, SingleWriter = false }); + + public void Enqueue(JobEntity job) + { + lock (_lock) + { + _pq.Enqueue(job, (int)job.Priority); + } + // Always write to signal channel to notify consumers. + _ = _signal.Writer.WriteAsync(job); + } + + public async Task DequeueAsync(CancellationToken cancellationToken) + { + // Wait until a job is signaled and then pop from our priority queue. + try + { + await _signal.Reader.ReadAsync(cancellationToken); + } + catch (OperationCanceledException) { return null; } + + lock (_lock) + { + if (_pq.TryDequeue(out var job, out _)) + { + return job; + } + } + + return null; + } + } +} \ No newline at end of file diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs b/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs new file mode 100644 index 00000000..0903c846 --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs @@ -0,0 +1,278 @@ +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.DependencyInjection; +using CrispyWaffle.BackgroundJobs.Abstractions; +using System.Text.Json; + +namespace CrispyWaffle.BackgroundJobs.Core +{ + public class BackgroundWorker : BackgroundService + { + private readonly IServiceProvider _provider; + private readonly BackgroundJobQueue? _queue; + private readonly IJobStore? _store; + private readonly IJobHandlerRegistry _registry; + private readonly JobOptions _options; + private readonly ILogger _logger; + + public BackgroundWorker(IServiceProvider provider, JobOptions options, ILogger logger) + { + _provider = provider; + _options = options; + _logger = logger; + + // These may or may not be registered depending on configuration; resolve lazily per scope + _queue = provider.GetService(); + _store = provider.GetService(); + _registry = provider.GetRequiredService(); + } + + protected override Task ExecuteAsync(CancellationToken stoppingToken) + { + _logger.LogInformation("BackgroundWorker starting with {Workers} workers", _options.WorkerCount); + + var tasks = new Task[_options.WorkerCount]; + for (int i = 0; i < _options.WorkerCount; i++) tasks[i] = Task.Run(() => WorkerLoopAsync(i, stoppingToken), stoppingToken); + + return Task.WhenAll(tasks); + } + + private async Task WorkerLoopAsync(int workerId, CancellationToken stoppingToken) + { + _logger.LogInformation("Worker {WorkerId} started", workerId); + + while (!stoppingToken.IsCancellationRequested) + { + try + { + JobEntity? job = null; + + if (_store != null) + { + job = await _store.FetchNextAsync(stoppingToken); + if (job == null) + { + await Task.Delay(_options.PollingInterval, stoppingToken); + continue; + } + } + else if (_queue != null) + { + job = await _queue.DequeueAsync(stoppingToken); + if (job == null) + { + await Task.Delay(TimeSpan.FromMilliseconds(200), stoppingToken); + continue; + } + } + else + { + _logger.LogWarning("No job queue nor store configured. Worker sleeping."); + await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken); + continue; + } + + if (job == null) continue; + + // If job has ScheduledAt in future, re-enqueue / re-schedule. + if (job.ScheduledAt.HasValue && job.ScheduledAt.Value > DateTimeOffset.UtcNow) + { + // For persistence, mark back to pending with same ScheduledAt; for in-memory, just re-enqueue. + if (_store != null) + { + // mark as Pending again (store.FetchNextAsync should have returned only due jobs, this is defensive) + await _store.MarkRetryAsync(job.Id, job.ScheduledAt, job.Attempt, stoppingToken); + } + else + { + _queue!.Enqueue(job); + } + + continue; + } + + // Process job with scope so handlers can have scoped services. + using var scope = _provider.CreateScope(); + var scopedProvider = scope.ServiceProvider; + + var success = false; + try + { + success = await ProcessJobAsync(job, scopedProvider, stoppingToken); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + // If shutting down, put the job back to pending (or leave DB in Processing if store does locking; here we mark retry) + if (_store != null) + { + var next = DateTimeOffset.UtcNow.AddSeconds(5); + await _store.MarkRetryAsync(job.Id, next, job.Attempt, stoppingToken); + } + else + { + _queue!.Enqueue(job); + } + break; // exit loop to allow graceful shutdown + } + + if (!success) + { + _logger.LogWarning("Job {JobId} failed after processing attempt {Attempt}", job.Id, job.Attempt); + } + } + catch (Exception ex) + { + _logger.LogError(ex, "Worker loop encountered an unexpected error"); + await Task.Delay(TimeSpan.FromSeconds(1), stoppingToken); + } + } + + _logger.LogInformation("Worker {WorkerId} stopping", workerId); + } + + private async Task ProcessJobAsync(JobEntity job, IServiceProvider scopedProvider, CancellationToken cancellationToken) + { + try + { + if (!_registry.TryGet(job.HandlerName, out var handlerType, out var dataType)) + { + var msg = $"No handler registered for '{job.HandlerName}'"; + _logger.LogError(msg); + if (_store != null) await _store.MarkFailedAsync(job.Id, msg, cancellationToken); + return false; + } + + // Deserialize payload into dataType + object? data = null; + if (!string.IsNullOrWhiteSpace(job.Payload)) + { + data = JsonSerializer.Deserialize(job.Payload, dataType ?? typeof(object)); + } + + // Resolve handler instance + var handler = scopedProvider.GetService(handlerType!); + if (handler == null) + { + var msg = $"Handler type {handlerType} not resolved from DI"; + _logger.LogError(msg); + if (_store != null) await _store.MarkFailedAsync(job.Id, msg, cancellationToken); + return false; + } + + // Invoke handler using reflection and await the Task + var result = await InvokeHandleAsync(handler, data, cancellationToken); + + if (result.Success) + { + if (_store != null) await _store.MarkCompletedAsync(job.Id, cancellationToken); + return true; + } + + // handle retry + job.Attempt++; + var shouldRetry = result.Retry && job.Attempt < job.MaxAttempt; + if (shouldRetry) + { + var backoffSeconds = ComputeBackoffSeconds(job.Attempt); + var nextAttempt = DateTimeOffset.UtcNow.AddSeconds(backoffSeconds); + _logger.LogInformation("Scheduling retry for job {JobId} after {Delay}s (attempt {Attempt}/{MaxAttempt})", job.Id, backoffSeconds, job.Attempt, job.MaxAttempt); + + if (_store != null) + { + await _store.MarkRetryAsync(job.Id, nextAttempt, job.Attempt, cancellationToken); + } + else + { + job.ScheduledAt = nextAttempt; + _queue!.Enqueue(job); + } + } + else + { + // exhaust + var err = result.ErrorMessage ?? "Job failed"; + if (_store != null) await _store.MarkFailedAsync(job.Id, err, cancellationToken); + else _logger.LogError("Job {JobId} failed and will not be retried: {Error}", job.Id, err); + } + + return false; + } + catch (Exception ex) + { + _logger.LogError(ex, "Exception when processing job {JobId}", job.Id); + if (_store != null) + { + // schedule retry with backoff + job.Attempt++; + var nextAttempt = DateTimeOffset.UtcNow.AddSeconds(ComputeBackoffSeconds(job.Attempt)); + await _store.MarkRetryAsync(job.Id, nextAttempt, job.Attempt, cancellationToken); + } + else + { + job.Attempt++; + job.ScheduledAt = DateTimeOffset.UtcNow.AddSeconds(ComputeBackoffSeconds(job.Attempt)); + _queue!.Enqueue(job); + } + + return false; + } + } + + private static int ComputeBackoffSeconds(int attempt) + { + // exponential backoff with cap + var seconds = (int)Math.Pow(2, Math.Min(attempt, 10)); + return Math.Min(seconds, 300); + } + + private static async Task InvokeHandleAsync(object handler, object? data, CancellationToken cancellationToken) + { + // Find HandleAsync method (TData, CancellationToken) + var method = handler.GetType().GetMethod("HandleAsync"); + if (method == null) throw new InvalidOperationException("Handler does not implement HandleAsync"); + + var parameters = method.GetParameters(); + object? invokeArg1 = null; + object? invokeArg2 = null; + if (parameters.Length == 2) + { + invokeArg1 = data; + invokeArg2 = cancellationToken; + } + else if (parameters.Length == 1) + { + invokeArg1 = data ?? cancellationToken; + if (invokeArg1 is CancellationToken) + { + invokeArg2 = null; + } + } + + var taskObj = method.Invoke(handler, invokeArg2 == null ? new[] { invokeArg1 } : new[] { invokeArg1, invokeArg2 }); + if (taskObj == null) throw new InvalidOperationException("Handler returned null instead of Task"); + + // support Task and Task + if (taskObj is Task typed) + { + return await typed; + } + + if (taskObj is Task task) + { + await task; + // attempt to read Result property (in case of Task) + var resultProperty = task.GetType().GetProperty("Result"); + if (resultProperty != null) + { + var res = resultProperty.GetValue(task); + if (res is JobResult jr) return jr; + } + + // else assume success + return JobResult.Ok(); + } + + throw new InvalidOperationException("Handler.HandleAsync returned unsupported type"); + } + } +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/IJobHandlerRegistry.cs b/Src/CrispyWaffle.BackgroundJobs/Core/IJobHandlerRegistry.cs new file mode 100644 index 00000000..f589ba5a --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Core/IJobHandlerRegistry.cs @@ -0,0 +1,10 @@ +namespace CrispyWaffle.BackgroundJobs.Core +{ + + public interface IJobHandlerRegistry + { + void Register(string handlerName) where THandler : class, CrispyWaffle.BackgroundJobs.Abstractions.IBackgroundJobHandler; + + bool TryGet(string handlerName, out Type? handlerType, out Type? dataType); + } +} \ No newline at end of file diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/JobDispatcher.cs b/Src/CrispyWaffle.BackgroundJobs/Core/JobDispatcher.cs new file mode 100644 index 00000000..df070132 --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Core/JobDispatcher.cs @@ -0,0 +1,77 @@ +using System.Text.Json; +using CrispyWaffle.BackgroundJobs.Abstractions; + +namespace CrispyWaffle.BackgroundJobs.Core +{ + public class JobDispatcher + { + private readonly IServiceProvider _provider; + private readonly BackgroundJobQueue? _queue; + private readonly IJobStore? _store; + private readonly JobOptions _options; + + public JobDispatcher(IServiceProvider provider, BackgroundJobQueue? queue, IJobStore? store, JobOptions options) + { + _provider = provider; + _queue = queue; + _store = store; + _options = options; + } + + public async Task EnqueueAsync(string handlerName, object payload, int? maxAttempts = null, JobPriority priority = JobPriority.Normal) + { + var job = new JobEntity + { + HandlerName = handlerName, + Payload = JsonSerializer.Serialize(payload), + Priority = priority, + MaxAttempt = maxAttempts ?? _options.DefaultMaxAttempt, + Status = JobStatus.Pending, + CreatedAt = DateTimeOffset.UtcNow + }; + + if (_store != null) + { + await _store.SaveAsync(job); + } + else + { + _queue?.Enqueue(job); + } + + return job.Id; + } + + public async Task ScheduleAsync(string handlerName, object payload, TimeSpan delay, int? maxAttempts = null, JobPriority priority = JobPriority.Normal) + { + var job = new JobEntity + { + HandlerName = handlerName, + Payload = JsonSerializer.Serialize(payload), + Priority = priority, + MaxAttempt = maxAttempts ?? _options.DefaultMaxAttempt, + Status = JobStatus.Pending, + ScheduledAt = DateTimeOffset.UtcNow.Add(delay), + CreatedAt = DateTimeOffset.UtcNow + }; + + if (_store != null) + { + await _store.SaveAsync(job); + } + else + { + // schedule in-memory using Task.Delay + _ = Task.Run(async () => + { + try + { + await Task.Delay(delay); + _queue?.Enqueue(job); + } + catch { /* suppressed */ } + }); + } + } + } +} \ No newline at end of file diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/JobHandlerRegistry.cs b/Src/CrispyWaffle.BackgroundJobs/Core/JobHandlerRegistry.cs new file mode 100644 index 00000000..af0ff4c4 --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Core/JobHandlerRegistry.cs @@ -0,0 +1,29 @@ +using System.Collections.Concurrent; + +namespace CrispyWaffle.BackgroundJobs.Core +{ + public class JobHandlerRegistry : IJobHandlerRegistry + { + private readonly ConcurrentDictionary _map = new(); + + public void Register(string handlerName) where THandler : class, CrispyWaffle.BackgroundJobs.Abstractions.IBackgroundJobHandler + { + if (string.IsNullOrWhiteSpace(handlerName)) throw new ArgumentException("handlerName required", nameof(handlerName)); + _map[handlerName] = (typeof(THandler), typeof(TData)); + } + + public bool TryGet(string handlerName, out Type? handlerType, out Type? dataType) + { + if (handlerName != null && _map.TryGetValue(handlerName, out var pair)) + { + handlerType = pair.handlerType; + dataType = pair.dataType; + return true; + } + + handlerType = null; + dataType = null; + return false; + } + } +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/JobOptions.cs b/Src/CrispyWaffle.BackgroundJobs/Core/JobOptions.cs new file mode 100644 index 00000000..4f6ce597 --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Core/JobOptions.cs @@ -0,0 +1,26 @@ +namespace CrispyWaffle.BackgroundJobs.Core +{ + + public class JobOptions + { + /// + /// Number of concurrent worker loops. + /// + public int WorkerCount { get; set; } = Environment.ProcessorCount; + + /// + /// Default maximum retry attempts for jobs enqueued without explicit MaxAttempt. + /// + public int DefaultMaxAttempt { get; set; } = 3; + + /// + /// Base polling delay (when using persistent store and no job found). + /// + public TimeSpan PollingInterval { get; set; } = TimeSpan.FromSeconds(2); + + /// + /// Maximum exponential backoff seconds for retry. + /// + public int MaxBackoffSeconds { get; set; } = 300; // 5 minutes + } +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/JobResult.cs b/Src/CrispyWaffle.BackgroundJobs/Core/JobResult.cs new file mode 100644 index 00000000..4d2e4166 --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Core/JobResult.cs @@ -0,0 +1,8 @@ +namespace CrispyWaffle.BackgroundJobs.Core +{ + public sealed record JobResult(bool Success, bool Retry = false, string? ErrorMessage = null) + { + public static JobResult Ok() => new(true); + public static JobResult Fail(string message, bool retry = false) => new(false, retry, message); + } +} diff --git a/Src/CrispyWaffle.BackgroundJobs/CrispyWaffle.BackgroundJobs.csproj b/Src/CrispyWaffle.BackgroundJobs/CrispyWaffle.BackgroundJobs.csproj new file mode 100644 index 00000000..11c40e78 --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/CrispyWaffle.BackgroundJobs.csproj @@ -0,0 +1,20 @@ + + + + net8.0 + enable + enable + + + + + + + + + + + + + + diff --git a/Src/CrispyWaffle.BackgroundJobs/Exceptions/JobFailedException.cs b/Src/CrispyWaffle.BackgroundJobs/Exceptions/JobFailedException.cs new file mode 100644 index 00000000..0e47ff1a --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Exceptions/JobFailedException.cs @@ -0,0 +1,7 @@ +namespace CrispyWaffle.BackgroundJobs.Exceptions +{ + public class JobFailedException : Exception + { + public JobFailedException(string message) : base(message) { } + } +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Extensions/ServiceCollectionExtensions.cs b/Src/CrispyWaffle.BackgroundJobs/Extensions/ServiceCollectionExtensions.cs new file mode 100644 index 00000000..726eb9ba --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Extensions/ServiceCollectionExtensions.cs @@ -0,0 +1,90 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using CrispyWaffle.BackgroundJobs.Core; +using CrispyWaffle.BackgroundJobs.Abstractions; +using CrispyWaffle.BackgroundJobs.Persistence; +using CrispyWaffle.BackgroundJobs.Monitoring; +using Microsoft.AspNetCore.Hosting; + +namespace CrispyWaffle.BackgroundJobs.Extensions +{ + public static class ServiceCollectionExtensions + { + /// + /// Register core background job services. By default uses InMemoryJobStore. To use EF Core, call AddDbContext before this and pass useEfCore=true. + /// + public static IServiceCollection AddCrispyBackgroundJobs(this IServiceCollection services, Action? configureOptions = null, bool useEfCoreStore = false) + { + var options = new JobOptions(); + configureOptions?.Invoke(options); + + services.AddSingleton(options); + services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(); + + // Register appropriate store + if (useEfCoreStore) + { + // JobDbContext must be registered by caller + services.TryAddScoped(); + } + else + { + services.TryAddSingleton(); + services.TryAddSingleton(); + } + + // Scheduler & worker + services.AddSingleton(); + + // Hosted service (worker) + services.AddHostedService(); + + return services; + } + + /// + /// Register a typed job handler where handler is resolved from DI and payload is serialized as JSON. + /// handlerName is the logical identifier used when enqueueing jobs. + /// + public static IServiceCollection AddJobHandler(this IServiceCollection services, string handlerName) + where THandler : class, IBackgroundJobHandler + { + services.AddScoped(); + // The registry registration happens at runtime; to ensure the registry has the entry we add a post-processor + services.AddSingleton(sp => new JobHandlerStartupFilter(sp, handlerName, typeof(THandler), typeof(TData))); + return services; + } + + // StartupFilter ensures the registry is populated after DI container is built + private class JobHandlerStartupFilter : Microsoft.AspNetCore.Hosting.IStartupFilter + { + private readonly IServiceProvider _sp; + private readonly string _handlerName; + private readonly Type _handlerType; + private readonly Type _dataType; + + public JobHandlerStartupFilter(IServiceProvider sp, string handlerName, Type handlerType, Type dataType) + { + _sp = sp; _handlerName = handlerName; _handlerType = handlerType; _dataType = dataType; + } + + public Action Configure(Action next) + { + return app => + { + var registry = _sp.GetRequiredService(); + var registerMethod = registry.GetType().GetMethod("Register")?.MakeGenericMethod(_handlerType, _dataType); + // If Register(string) exists we call via reflection + if (registerMethod != null) + { + registerMethod.Invoke(registry, new object[] { _handlerName }); + } + + next(app); + }; + } + } + } +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Monitoring/JobMetrics.cs b/Src/CrispyWaffle.BackgroundJobs/Monitoring/JobMetrics.cs new file mode 100644 index 00000000..58241216 --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Monitoring/JobMetrics.cs @@ -0,0 +1,19 @@ +using System.Collections.Concurrent; + +namespace CrispyWaffle.BackgroundJobs.Monitoring +{ + public class JobMetrics + { + private readonly ConcurrentDictionary _counters = new(); + + public void Increment(string key) + { + _counters.AddOrUpdate(key, 1, (_, v) => v + 1); + } + + public long Get(string key) => _counters.TryGetValue(key, out var v) ? v : 0; + + public System.Collections.Generic.IDictionary Snapshot() => new System.Collections.Generic.Dictionary(_counters); + } +} + diff --git a/Src/CrispyWaffle.BackgroundJobs/Monitoring/MonitoringMiddleware.cs b/Src/CrispyWaffle.BackgroundJobs/Monitoring/MonitoringMiddleware.cs new file mode 100644 index 00000000..e4cd9c19 --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Monitoring/MonitoringMiddleware.cs @@ -0,0 +1,29 @@ +using System.Text.Json; +using Microsoft.AspNetCore.Http; + +namespace CrispyWaffle.BackgroundJobs.Monitoring +{ + public class MonitoringMiddleware + { + private readonly RequestDelegate _next; + private readonly JobMetrics _metrics; + + public MonitoringMiddleware(RequestDelegate next, JobMetrics metrics) + { + _next = next; + _metrics = metrics; + } + + public async Task InvokeAsync(HttpContext context) + { + if (context.Request.Path.StartsWithSegments("/jobs/metrics")) + { + context.Response.ContentType = "application/json"; + await context.Response.WriteAsync(JsonSerializer.Serialize(_metrics.Snapshot())); + return; + } + + await _next(context); + } + } +} \ No newline at end of file diff --git a/Src/CrispyWaffle.BackgroundJobs/Persistence/EfCoreJobStore.cs b/Src/CrispyWaffle.BackgroundJobs/Persistence/EfCoreJobStore.cs new file mode 100644 index 00000000..74a2c7c3 --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Persistence/EfCoreJobStore.cs @@ -0,0 +1,86 @@ +using CrispyWaffle.BackgroundJobs.Abstractions; +using Microsoft.EntityFrameworkCore; + +namespace CrispyWaffle.BackgroundJobs.Persistence +{ + public class EfCoreJobStore : IJobStore + { + private readonly JobDbContext _db; + + public EfCoreJobStore(JobDbContext db) + { + _db = db; + } + + public async Task SaveAsync(JobEntity job, CancellationToken cancellationToken = default) + { + job.UpdatedAt = DateTimeOffset.UtcNow; + _db.Jobs.Add(job); + await _db.SaveChangesAsync(cancellationToken); + } + + public async Task FetchNextAsync(CancellationToken cancellationToken = default) + { + // We attempt to find a pending job that is due, ordered by priority and created time. We then mark it Processing inside a transaction + var now = DateTimeOffset.UtcNow; + + // This uses a simple approach: select candidate IDs then try to update one row in a transaction; this reduces race conditions. + var candidate = await _db.Jobs + .Where(j => j.Status == JobStatus.Pending && (j.ScheduledAt == null || j.ScheduledAt <= now)) + .OrderBy(j => (int)j.Priority) + .ThenBy(j => j.CreatedAt) + .Select(j => j.Id) + .FirstOrDefaultAsync(cancellationToken); + + if (candidate == Guid.Empty) return null; + + // Try to claim the job + var job = await _db.Jobs.FirstOrDefaultAsync(j => j.Id == candidate, cancellationToken); + if (job == null) return null; + + job.Status = JobStatus.Processing; + job.UpdatedAt = DateTimeOffset.UtcNow; + + try + { + await _db.SaveChangesAsync(cancellationToken); + return job; + } + catch (DbUpdateConcurrencyException) + { + // Someone else claimed it + return null; + } + } + + public async Task MarkCompletedAsync(Guid jobId, CancellationToken cancellationToken = default) + { + var job = await _db.Jobs.FirstOrDefaultAsync(j => j.Id == jobId, cancellationToken); + if (job == null) return; + job.Status = JobStatus.Completed; + job.UpdatedAt = DateTimeOffset.UtcNow; + await _db.SaveChangesAsync(cancellationToken); + } + + public async Task MarkFailedAsync(Guid jobId, string error, CancellationToken cancellationToken = default) + { + var job = await _db.Jobs.FirstOrDefaultAsync(j => j.Id == jobId, cancellationToken); + if (job == null) return; + job.Status = JobStatus.Failed; + job.LastError = error; + job.UpdatedAt = DateTimeOffset.UtcNow; + await _db.SaveChangesAsync(cancellationToken); + } + + public async Task MarkRetryAsync(Guid jobId, DateTimeOffset? nextAttemptAt, int attemptCount, CancellationToken cancellationToken = default) + { + var job = await _db.Jobs.FirstOrDefaultAsync(j => j.Id == jobId, cancellationToken); + if (job == null) return; + job.Attempt = attemptCount; + job.Status = JobStatus.Pending; + job.ScheduledAt = nextAttemptAt; + job.UpdatedAt = DateTimeOffset.UtcNow; + await _db.SaveChangesAsync(cancellationToken); + } + } +} \ No newline at end of file diff --git a/Src/CrispyWaffle.BackgroundJobs/Persistence/InMemoryJobStore.cs b/Src/CrispyWaffle.BackgroundJobs/Persistence/InMemoryJobStore.cs new file mode 100644 index 00000000..c11a8d4e --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Persistence/InMemoryJobStore.cs @@ -0,0 +1,74 @@ +using System.Collections.Concurrent; +using CrispyWaffle.BackgroundJobs.Abstractions; + +namespace CrispyWaffle.BackgroundJobs.Persistence +{ + public class InMemoryJobStore : IJobStore + { + private readonly ConcurrentDictionary _storage = new(); + + public Task SaveAsync(JobEntity job, CancellationToken cancellationToken = default) + { + job.UpdatedAt = DateTimeOffset.UtcNow; + _storage[job.Id] = job; + return Task.CompletedTask; + } + + public Task FetchNextAsync(CancellationToken cancellationToken = default) + { + var now = DateTimeOffset.UtcNow; + var next = _storage.Values + .Where(j => j.Status == JobStatus.Pending && (!j.ScheduledAt.HasValue || j.ScheduledAt.Value <= now)) + .OrderBy(j => (int)j.Priority) + .ThenBy(j => j.CreatedAt) + .FirstOrDefault(); + + if (next != null) + { + // try to transition to Processing; using atomic update + next.Status = JobStatus.Processing; + next.UpdatedAt = DateTimeOffset.UtcNow; + _storage[next.Id] = next; + return Task.FromResult(next); + } + + return Task.FromResult(null); + } + + public Task MarkCompletedAsync(Guid jobId, CancellationToken cancellationToken = default) + { + if (_storage.TryGetValue(jobId, out var job)) + { + job.Status = JobStatus.Completed; + job.UpdatedAt = DateTimeOffset.UtcNow; + _storage[jobId] = job; + } + return Task.CompletedTask; + } + + public Task MarkFailedAsync(Guid jobId, string error, CancellationToken cancellationToken = default) + { + if (_storage.TryGetValue(jobId, out var job)) + { + job.Status = JobStatus.Failed; + job.LastError = error; + job.UpdatedAt = DateTimeOffset.UtcNow; + _storage[jobId] = job; + } + return Task.CompletedTask; + } + + public Task MarkRetryAsync(Guid jobId, DateTimeOffset? nextAttemptAt, int attemptCount, CancellationToken cancellationToken = default) + { + if (_storage.TryGetValue(jobId, out var job)) + { + job.Attempt = attemptCount; + job.Status = JobStatus.Pending; + job.ScheduledAt = nextAttemptAt; + job.UpdatedAt = DateTimeOffset.UtcNow; + _storage[jobId] = job; + } + return Task.CompletedTask; + } + } +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Persistence/JobDbContext.cs b/Src/CrispyWaffle.BackgroundJobs/Persistence/JobDbContext.cs new file mode 100644 index 00000000..ee719ad4 --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Persistence/JobDbContext.cs @@ -0,0 +1,31 @@ +using System.Reflection.Emit; +using CrispyWaffle.BackgroundJobs.Abstractions; +using Microsoft.EntityFrameworkCore; + +namespace CrispyWaffle.BackgroundJobs.Persistence +{ + public class JobDbContext : DbContext + { + public JobDbContext(DbContextOptions options) : base(options) { } + + public DbSet Jobs { get; set; } = null!; + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + base.OnModelCreating(modelBuilder); + + modelBuilder.Entity(eb => + { + eb.HasKey(j => j.Id); + eb.Property(j => j.HandlerName).IsRequired().HasMaxLength(256); + eb.Property(j => j.Payload).IsRequired(false); + eb.Property(j => j.Priority).HasConversion(); + eb.Property(j => j.Status).HasConversion(); + eb.Property(j => j.CreatedAt).IsRequired(); + eb.Property(j => j.UpdatedAt).IsRequired(); + eb.Property(j => j.MaxAttempt).HasDefaultValue(3); + eb.Property(j => j.Attempt).HasDefaultValue(0); + }); + } + } +} \ No newline at end of file diff --git a/Src/CrispyWaffle.BackgroundJobs/Scheduling/CronJobScheduler.cs b/Src/CrispyWaffle.BackgroundJobs/Scheduling/CronJobScheduler.cs new file mode 100644 index 00000000..b8acaa15 --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Scheduling/CronJobScheduler.cs @@ -0,0 +1,40 @@ +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; + +namespace CrispyWaffle.BackgroundJobs.Scheduling +{ + /// + /// CronJobScheduler is a lightweight helper that triggers an action on a CRON expression. + /// Note: to support CRON syntax you'd typically add a dependency such as NCrontab or Cronos. + /// This is a stub showing how to wire a hosted service cron runner. + /// + public abstract class CronJobService : BackgroundService + { + private readonly ILogger _logger; + + protected CronJobService(ILogger logger) + { + _logger = logger; + } + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + // Example stub: run every minute + while (!stoppingToken.IsCancellationRequested) + { + try + { + await ExecuteOnce(stoppingToken); + } + catch (Exception ex) + { + _logger.LogError(ex, "Cron job execution error"); + } + + await Task.Delay(TimeSpan.FromMinutes(1), stoppingToken); + } + } + + protected abstract Task ExecuteOnce(CancellationToken stoppingToken); + } +} \ No newline at end of file diff --git a/Src/CrispyWaffle.BackgroundJobs/Scheduling/DelayedJobScheduler.cs b/Src/CrispyWaffle.BackgroundJobs/Scheduling/DelayedJobScheduler.cs new file mode 100644 index 00000000..02194a82 --- /dev/null +++ b/Src/CrispyWaffle.BackgroundJobs/Scheduling/DelayedJobScheduler.cs @@ -0,0 +1,31 @@ +using CrispyWaffle.BackgroundJobs.Abstractions; +using CrispyWaffle.BackgroundJobs.Core; +using Microsoft.Extensions.DependencyInjection; + +namespace CrispyWaffle.BackgroundJobs.Scheduling +{ + /// + /// Schedules jobs by either creating a persisted scheduled job (if store available) or using in-memory Task.Delay. + /// + public class DelayedJobScheduler : IJobScheduler + { + private readonly IServiceProvider _provider; + + public DelayedJobScheduler(IServiceProvider provider) + { + _provider = provider; + } + + public Task EnqueueAsync(string handlerName, object payload, int maxAttempts = 3, JobPriority priority = JobPriority.Normal) + { + var dispatcher = _provider.GetRequiredService(); + return dispatcher.EnqueueAsync(handlerName, payload, maxAttempts, priority); + } + + public Task ScheduleAsync(string handlerName, object payload, TimeSpan delay, int maxAttempts = 3, JobPriority priority = JobPriority.Normal) + { + var dispatcher = _provider.GetRequiredService(); + return dispatcher.ScheduleAsync(handlerName, payload, delay, maxAttempts, priority); + } + } +} \ No newline at end of file From 5e3ddd49466ea942983707e3407d1fdb77f66395 Mon Sep 17 00:00:00 2001 From: codefactor-io Date: Sat, 16 Aug 2025 07:04:14 +0000 Subject: [PATCH 07/10] [CodeFactor] Apply fixes --- Src/CrispyWaffle.BackgroundJobs/Core/IJobHandlerRegistry.cs | 3 +-- Src/CrispyWaffle.BackgroundJobs/Core/JobOptions.cs | 3 +-- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/IJobHandlerRegistry.cs b/Src/CrispyWaffle.BackgroundJobs/Core/IJobHandlerRegistry.cs index f589ba5a..4e52d024 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Core/IJobHandlerRegistry.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Core/IJobHandlerRegistry.cs @@ -1,6 +1,5 @@ -namespace CrispyWaffle.BackgroundJobs.Core +namespace CrispyWaffle.BackgroundJobs.Core { - public interface IJobHandlerRegistry { void Register(string handlerName) where THandler : class, CrispyWaffle.BackgroundJobs.Abstractions.IBackgroundJobHandler; diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/JobOptions.cs b/Src/CrispyWaffle.BackgroundJobs/Core/JobOptions.cs index 4f6ce597..d3e3fd2c 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Core/JobOptions.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Core/JobOptions.cs @@ -1,6 +1,5 @@ -namespace CrispyWaffle.BackgroundJobs.Core +namespace CrispyWaffle.BackgroundJobs.Core { - public class JobOptions { /// From 2c151477e01490b16950f5e04db5303c960e2485 Mon Sep 17 00:00:00 2001 From: Guilherme Branco Stracini Date: Sat, 4 Jul 2026 22:50:25 +0100 Subject: [PATCH 08/10] style: improve code readability and consistency +semver: minor Reformat multiline interface and class methods to enhance code readability. Align method parameters across multiple lines consistently in all files. Add missing newline characters at the end of files for better standards compliance. These changes improve maintainability without altering functionality, ensuring code consistency across the project. --- Directory.Packages.props | 7 +- .../Abstractions/IBackgroundJob.cs | 2 +- .../Abstractions/IJobScheduler.cs | 17 ++- .../Abstractions/IJobStore.cs | 15 ++- .../Abstractions/JobEntity.cs | 2 +- .../Abstractions/JobPriority.cs | 2 +- .../Abstractions/JobStatus.cs | 4 +- .../Core/BackgroundJobQueue.cs | 15 ++- .../Core/BackgroundWorker.cs | 115 ++++++++++++++---- .../Core/IJobHandlerRegistry.cs | 6 +- .../Core/JobDispatcher.cs | 32 +++-- .../Core/JobHandlerRegistry.cs | 10 +- .../Core/JobResult.cs | 4 +- .../CrispyWaffle.BackgroundJobs.csproj | 2 - .../Exceptions/JobFailedException.cs | 3 +- .../Extensions/ServiceCollectionExtensions.cs | 52 ++++++-- .../Monitoring/JobMetrics.cs | 4 +- .../Monitoring/MonitoringMiddleware.cs | 2 +- .../Persistence/EfCoreJobStore.cs | 41 +++++-- .../Persistence/InMemoryJobStore.cs | 20 ++- .../Persistence/JobDbContext.cs | 5 +- .../Scheduling/CronJobScheduler.cs | 2 +- .../Scheduling/DelayedJobScheduler.cs | 17 ++- 23 files changed, 283 insertions(+), 96 deletions(-) diff --git a/Directory.Packages.props b/Directory.Packages.props index c14e07e3..2ce4b9f3 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -18,7 +18,10 @@ - + @@ -42,4 +45,4 @@ runtime; build; native; contentfiles; analyzers; buildtransitive - \ No newline at end of file + diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/IBackgroundJob.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IBackgroundJob.cs index cc4562a9..eb72223d 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Abstractions/IBackgroundJob.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IBackgroundJob.cs @@ -10,4 +10,4 @@ public interface IBackgroundJob { Task ExecuteAsync(CancellationToken cancellationToken); } -} \ No newline at end of file +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobScheduler.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobScheduler.cs index f01a9d4a..5b4d777a 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobScheduler.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobScheduler.cs @@ -2,8 +2,19 @@ { public interface IJobScheduler { - Task ScheduleAsync(string handlerName, object payload, TimeSpan delay, int maxAttempts = 3, JobPriority priority = JobPriority.Normal); + Task ScheduleAsync( + string handlerName, + object payload, + TimeSpan delay, + int maxAttempts = 3, + JobPriority priority = JobPriority.Normal + ); - Task EnqueueAsync(string handlerName, object payload, int maxAttempts = 3, JobPriority priority = JobPriority.Normal); + Task EnqueueAsync( + string handlerName, + object payload, + int maxAttempts = 3, + JobPriority priority = JobPriority.Normal + ); } -} \ No newline at end of file +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobStore.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobStore.cs index 7ba99e15..aeb4d420 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobStore.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobStore.cs @@ -12,8 +12,17 @@ public interface IJobStore Task MarkCompletedAsync(Guid jobId, CancellationToken cancellationToken = default); - Task MarkFailedAsync(Guid jobId, string error, CancellationToken cancellationToken = default); + Task MarkFailedAsync( + Guid jobId, + string error, + CancellationToken cancellationToken = default + ); - Task MarkRetryAsync(Guid jobId, DateTimeOffset? nextAttemptAt, int attemptCount, CancellationToken cancellationToken = default); + Task MarkRetryAsync( + Guid jobId, + DateTimeOffset? nextAttemptAt, + int attemptCount, + CancellationToken cancellationToken = default + ); } -} \ No newline at end of file +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobEntity.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobEntity.cs index bf97626f..1d09431b 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobEntity.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobEntity.cs @@ -44,4 +44,4 @@ public class JobEntity /// public int RetryDelaySeconds { get; set; } = 0; } -} \ No newline at end of file +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobPriority.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobPriority.cs index 621587ae..1ac54c2e 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobPriority.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobPriority.cs @@ -4,6 +4,6 @@ public enum JobPriority { High = 0, Normal = 1, - Low = 2 + Low = 2, } } diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobStatus.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobStatus.cs index c7ef22d9..8854e13a 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobStatus.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobStatus.cs @@ -6,6 +6,6 @@ public enum JobStatus Processing = 1, Completed = 2, Failed = 3, - Dead = 4 + Dead = 4, } -} \ No newline at end of file +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundJobQueue.cs b/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundJobQueue.cs index a1d01af2..b735e2d2 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundJobQueue.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundJobQueue.cs @@ -1,5 +1,5 @@ -using CrispyWaffle.BackgroundJobs.Abstractions; -using System.Threading.Channels; +using System.Threading.Channels; +using CrispyWaffle.BackgroundJobs.Abstractions; namespace CrispyWaffle.BackgroundJobs.Core { @@ -12,7 +12,9 @@ public class BackgroundJobQueue { private readonly object _lock = new(); private readonly PriorityQueue _pq = new(); - private readonly Channel _signal = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = false, SingleWriter = false }); + private readonly Channel _signal = Channel.CreateUnbounded( + new UnboundedChannelOptions { SingleReader = false, SingleWriter = false } + ); public void Enqueue(JobEntity job) { @@ -31,7 +33,10 @@ public void Enqueue(JobEntity job) { await _signal.Reader.ReadAsync(cancellationToken); } - catch (OperationCanceledException) { return null; } + catch (OperationCanceledException) + { + return null; + } lock (_lock) { @@ -44,4 +49,4 @@ public void Enqueue(JobEntity job) return null; } } -} \ No newline at end of file +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs b/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs index 0903c846..d0a53871 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs @@ -1,8 +1,8 @@ -using Microsoft.Extensions.Hosting; -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.DependencyInjection; +using System.Text.Json; using CrispyWaffle.BackgroundJobs.Abstractions; -using System.Text.Json; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; namespace CrispyWaffle.BackgroundJobs.Core { @@ -15,7 +15,11 @@ public class BackgroundWorker : BackgroundService private readonly JobOptions _options; private readonly ILogger _logger; - public BackgroundWorker(IServiceProvider provider, JobOptions options, ILogger logger) + public BackgroundWorker( + IServiceProvider provider, + JobOptions options, + ILogger logger + ) { _provider = provider; _options = options; @@ -29,10 +33,14 @@ public BackgroundWorker(IServiceProvider provider, JobOptions options, ILogger WorkerLoopAsync(i, stoppingToken), stoppingToken); + for (int i = 0; i < _options.WorkerCount; i++) + tasks[i] = Task.Run(() => WorkerLoopAsync(i, stoppingToken), stoppingToken); return Task.WhenAll(tasks); } @@ -72,7 +80,8 @@ private async Task WorkerLoopAsync(int workerId, CancellationToken stoppingToken continue; } - if (job == null) continue; + if (job == null) + continue; // If job has ScheduledAt in future, re-enqueue / re-schedule. if (job.ScheduledAt.HasValue && job.ScheduledAt.Value > DateTimeOffset.UtcNow) @@ -81,7 +90,12 @@ private async Task WorkerLoopAsync(int workerId, CancellationToken stoppingToken if (_store != null) { // mark as Pending again (store.FetchNextAsync should have returned only due jobs, this is defensive) - await _store.MarkRetryAsync(job.Id, job.ScheduledAt, job.Attempt, stoppingToken); + await _store.MarkRetryAsync( + job.Id, + job.ScheduledAt, + job.Attempt, + stoppingToken + ); } else { @@ -117,7 +131,11 @@ private async Task WorkerLoopAsync(int workerId, CancellationToken stoppingToken if (!success) { - _logger.LogWarning("Job {JobId} failed after processing attempt {Attempt}", job.Id, job.Attempt); + _logger.LogWarning( + "Job {JobId} failed after processing attempt {Attempt}", + job.Id, + job.Attempt + ); } } catch (Exception ex) @@ -130,7 +148,11 @@ private async Task WorkerLoopAsync(int workerId, CancellationToken stoppingToken _logger.LogInformation("Worker {WorkerId} stopping", workerId); } - private async Task ProcessJobAsync(JobEntity job, IServiceProvider scopedProvider, CancellationToken cancellationToken) + private async Task ProcessJobAsync( + JobEntity job, + IServiceProvider scopedProvider, + CancellationToken cancellationToken + ) { try { @@ -138,7 +160,8 @@ private async Task ProcessJobAsync(JobEntity job, IServiceProvider scopedP { var msg = $"No handler registered for '{job.HandlerName}'"; _logger.LogError(msg); - if (_store != null) await _store.MarkFailedAsync(job.Id, msg, cancellationToken); + if (_store != null) + await _store.MarkFailedAsync(job.Id, msg, cancellationToken); return false; } @@ -155,7 +178,8 @@ private async Task ProcessJobAsync(JobEntity job, IServiceProvider scopedP { var msg = $"Handler type {handlerType} not resolved from DI"; _logger.LogError(msg); - if (_store != null) await _store.MarkFailedAsync(job.Id, msg, cancellationToken); + if (_store != null) + await _store.MarkFailedAsync(job.Id, msg, cancellationToken); return false; } @@ -164,7 +188,8 @@ private async Task ProcessJobAsync(JobEntity job, IServiceProvider scopedP if (result.Success) { - if (_store != null) await _store.MarkCompletedAsync(job.Id, cancellationToken); + if (_store != null) + await _store.MarkCompletedAsync(job.Id, cancellationToken); return true; } @@ -175,11 +200,22 @@ private async Task ProcessJobAsync(JobEntity job, IServiceProvider scopedP { var backoffSeconds = ComputeBackoffSeconds(job.Attempt); var nextAttempt = DateTimeOffset.UtcNow.AddSeconds(backoffSeconds); - _logger.LogInformation("Scheduling retry for job {JobId} after {Delay}s (attempt {Attempt}/{MaxAttempt})", job.Id, backoffSeconds, job.Attempt, job.MaxAttempt); + _logger.LogInformation( + "Scheduling retry for job {JobId} after {Delay}s (attempt {Attempt}/{MaxAttempt})", + job.Id, + backoffSeconds, + job.Attempt, + job.MaxAttempt + ); if (_store != null) { - await _store.MarkRetryAsync(job.Id, nextAttempt, job.Attempt, cancellationToken); + await _store.MarkRetryAsync( + job.Id, + nextAttempt, + job.Attempt, + cancellationToken + ); } else { @@ -191,8 +227,14 @@ private async Task ProcessJobAsync(JobEntity job, IServiceProvider scopedP { // exhaust var err = result.ErrorMessage ?? "Job failed"; - if (_store != null) await _store.MarkFailedAsync(job.Id, err, cancellationToken); - else _logger.LogError("Job {JobId} failed and will not be retried: {Error}", job.Id, err); + if (_store != null) + await _store.MarkFailedAsync(job.Id, err, cancellationToken); + else + _logger.LogError( + "Job {JobId} failed and will not be retried: {Error}", + job.Id, + err + ); } return false; @@ -204,13 +246,22 @@ private async Task ProcessJobAsync(JobEntity job, IServiceProvider scopedP { // schedule retry with backoff job.Attempt++; - var nextAttempt = DateTimeOffset.UtcNow.AddSeconds(ComputeBackoffSeconds(job.Attempt)); - await _store.MarkRetryAsync(job.Id, nextAttempt, job.Attempt, cancellationToken); + var nextAttempt = DateTimeOffset.UtcNow.AddSeconds( + ComputeBackoffSeconds(job.Attempt) + ); + await _store.MarkRetryAsync( + job.Id, + nextAttempt, + job.Attempt, + cancellationToken + ); } else { job.Attempt++; - job.ScheduledAt = DateTimeOffset.UtcNow.AddSeconds(ComputeBackoffSeconds(job.Attempt)); + job.ScheduledAt = DateTimeOffset.UtcNow.AddSeconds( + ComputeBackoffSeconds(job.Attempt) + ); _queue!.Enqueue(job); } @@ -225,11 +276,16 @@ private static int ComputeBackoffSeconds(int attempt) return Math.Min(seconds, 300); } - private static async Task InvokeHandleAsync(object handler, object? data, CancellationToken cancellationToken) + private static async Task InvokeHandleAsync( + object handler, + object? data, + CancellationToken cancellationToken + ) { // Find HandleAsync method (TData, CancellationToken) var method = handler.GetType().GetMethod("HandleAsync"); - if (method == null) throw new InvalidOperationException("Handler does not implement HandleAsync"); + if (method == null) + throw new InvalidOperationException("Handler does not implement HandleAsync"); var parameters = method.GetParameters(); object? invokeArg1 = null; @@ -248,8 +304,14 @@ private static async Task InvokeHandleAsync(object handler, object? d } } - var taskObj = method.Invoke(handler, invokeArg2 == null ? new[] { invokeArg1 } : new[] { invokeArg1, invokeArg2 }); - if (taskObj == null) throw new InvalidOperationException("Handler returned null instead of Task"); + var taskObj = method.Invoke( + handler, + invokeArg2 == null ? new[] { invokeArg1 } : new[] { invokeArg1, invokeArg2 } + ); + if (taskObj == null) + throw new InvalidOperationException( + "Handler returned null instead of Task" + ); // support Task and Task if (taskObj is Task typed) @@ -265,7 +327,8 @@ private static async Task InvokeHandleAsync(object handler, object? d if (resultProperty != null) { var res = resultProperty.GetValue(task); - if (res is JobResult jr) return jr; + if (res is JobResult jr) + return jr; } // else assume success diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/IJobHandlerRegistry.cs b/Src/CrispyWaffle.BackgroundJobs/Core/IJobHandlerRegistry.cs index 4e52d024..e3e27b8d 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Core/IJobHandlerRegistry.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Core/IJobHandlerRegistry.cs @@ -2,8 +2,10 @@ namespace CrispyWaffle.BackgroundJobs.Core { public interface IJobHandlerRegistry { - void Register(string handlerName) where THandler : class, CrispyWaffle.BackgroundJobs.Abstractions.IBackgroundJobHandler; + void Register(string handlerName) + where THandler : class, + CrispyWaffle.BackgroundJobs.Abstractions.IBackgroundJobHandler; bool TryGet(string handlerName, out Type? handlerType, out Type? dataType); } -} \ No newline at end of file +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/JobDispatcher.cs b/Src/CrispyWaffle.BackgroundJobs/Core/JobDispatcher.cs index df070132..6a8f407a 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Core/JobDispatcher.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Core/JobDispatcher.cs @@ -10,7 +10,12 @@ public class JobDispatcher private readonly IJobStore? _store; private readonly JobOptions _options; - public JobDispatcher(IServiceProvider provider, BackgroundJobQueue? queue, IJobStore? store, JobOptions options) + public JobDispatcher( + IServiceProvider provider, + BackgroundJobQueue? queue, + IJobStore? store, + JobOptions options + ) { _provider = provider; _queue = queue; @@ -18,7 +23,12 @@ public JobDispatcher(IServiceProvider provider, BackgroundJobQueue? queue, IJobS _options = options; } - public async Task EnqueueAsync(string handlerName, object payload, int? maxAttempts = null, JobPriority priority = JobPriority.Normal) + public async Task EnqueueAsync( + string handlerName, + object payload, + int? maxAttempts = null, + JobPriority priority = JobPriority.Normal + ) { var job = new JobEntity { @@ -27,7 +37,7 @@ public async Task EnqueueAsync(string handlerName, object payload, int? ma Priority = priority, MaxAttempt = maxAttempts ?? _options.DefaultMaxAttempt, Status = JobStatus.Pending, - CreatedAt = DateTimeOffset.UtcNow + CreatedAt = DateTimeOffset.UtcNow, }; if (_store != null) @@ -42,7 +52,13 @@ public async Task EnqueueAsync(string handlerName, object payload, int? ma return job.Id; } - public async Task ScheduleAsync(string handlerName, object payload, TimeSpan delay, int? maxAttempts = null, JobPriority priority = JobPriority.Normal) + public async Task ScheduleAsync( + string handlerName, + object payload, + TimeSpan delay, + int? maxAttempts = null, + JobPriority priority = JobPriority.Normal + ) { var job = new JobEntity { @@ -52,7 +68,7 @@ public async Task ScheduleAsync(string handlerName, object payload, TimeSpan del MaxAttempt = maxAttempts ?? _options.DefaultMaxAttempt, Status = JobStatus.Pending, ScheduledAt = DateTimeOffset.UtcNow.Add(delay), - CreatedAt = DateTimeOffset.UtcNow + CreatedAt = DateTimeOffset.UtcNow, }; if (_store != null) @@ -69,9 +85,11 @@ public async Task ScheduleAsync(string handlerName, object payload, TimeSpan del await Task.Delay(delay); _queue?.Enqueue(job); } - catch { /* suppressed */ } + catch + { /* suppressed */ + } }); } } } -} \ No newline at end of file +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/JobHandlerRegistry.cs b/Src/CrispyWaffle.BackgroundJobs/Core/JobHandlerRegistry.cs index af0ff4c4..606027a9 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Core/JobHandlerRegistry.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Core/JobHandlerRegistry.cs @@ -4,11 +4,15 @@ namespace CrispyWaffle.BackgroundJobs.Core { public class JobHandlerRegistry : IJobHandlerRegistry { - private readonly ConcurrentDictionary _map = new(); + private readonly ConcurrentDictionary _map = + new(); - public void Register(string handlerName) where THandler : class, CrispyWaffle.BackgroundJobs.Abstractions.IBackgroundJobHandler + public void Register(string handlerName) + where THandler : class, + CrispyWaffle.BackgroundJobs.Abstractions.IBackgroundJobHandler { - if (string.IsNullOrWhiteSpace(handlerName)) throw new ArgumentException("handlerName required", nameof(handlerName)); + if (string.IsNullOrWhiteSpace(handlerName)) + throw new ArgumentException("handlerName required", nameof(handlerName)); _map[handlerName] = (typeof(THandler), typeof(TData)); } diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/JobResult.cs b/Src/CrispyWaffle.BackgroundJobs/Core/JobResult.cs index 4d2e4166..d9a8531d 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Core/JobResult.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Core/JobResult.cs @@ -3,6 +3,8 @@ public sealed record JobResult(bool Success, bool Retry = false, string? ErrorMessage = null) { public static JobResult Ok() => new(true); - public static JobResult Fail(string message, bool retry = false) => new(false, retry, message); + + public static JobResult Fail(string message, bool retry = false) => + new(false, retry, message); } } diff --git a/Src/CrispyWaffle.BackgroundJobs/CrispyWaffle.BackgroundJobs.csproj b/Src/CrispyWaffle.BackgroundJobs/CrispyWaffle.BackgroundJobs.csproj index 11c40e78..6a747f0b 100644 --- a/Src/CrispyWaffle.BackgroundJobs/CrispyWaffle.BackgroundJobs.csproj +++ b/Src/CrispyWaffle.BackgroundJobs/CrispyWaffle.BackgroundJobs.csproj @@ -1,5 +1,4 @@  - net8.0 enable @@ -16,5 +15,4 @@ - diff --git a/Src/CrispyWaffle.BackgroundJobs/Exceptions/JobFailedException.cs b/Src/CrispyWaffle.BackgroundJobs/Exceptions/JobFailedException.cs index 0e47ff1a..210e7f87 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Exceptions/JobFailedException.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Exceptions/JobFailedException.cs @@ -2,6 +2,7 @@ { public class JobFailedException : Exception { - public JobFailedException(string message) : base(message) { } + public JobFailedException(string message) + : base(message) { } } } diff --git a/Src/CrispyWaffle.BackgroundJobs/Extensions/ServiceCollectionExtensions.cs b/Src/CrispyWaffle.BackgroundJobs/Extensions/ServiceCollectionExtensions.cs index 726eb9ba..55132b38 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Extensions/ServiceCollectionExtensions.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Extensions/ServiceCollectionExtensions.cs @@ -1,10 +1,10 @@ -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.DependencyInjection.Extensions; +using CrispyWaffle.BackgroundJobs.Abstractions; using CrispyWaffle.BackgroundJobs.Core; -using CrispyWaffle.BackgroundJobs.Abstractions; -using CrispyWaffle.BackgroundJobs.Persistence; using CrispyWaffle.BackgroundJobs.Monitoring; +using CrispyWaffle.BackgroundJobs.Persistence; using Microsoft.AspNetCore.Hosting; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; namespace CrispyWaffle.BackgroundJobs.Extensions { @@ -13,7 +13,11 @@ public static class ServiceCollectionExtensions /// /// Register core background job services. By default uses InMemoryJobStore. To use EF Core, call AddDbContext before this and pass useEfCore=true. /// - public static IServiceCollection AddCrispyBackgroundJobs(this IServiceCollection services, Action? configureOptions = null, bool useEfCoreStore = false) + public static IServiceCollection AddCrispyBackgroundJobs( + this IServiceCollection services, + Action? configureOptions = null, + bool useEfCoreStore = false + ) { var options = new JobOptions(); configureOptions?.Invoke(options); @@ -36,7 +40,10 @@ public static IServiceCollection AddCrispyBackgroundJobs(this IServiceCollection } // Scheduler & worker - services.AddSingleton(); + services.AddSingleton< + IJobScheduler, + CrispyWaffle.BackgroundJobs.Scheduling.DelayedJobScheduler + >(); // Hosted service (worker) services.AddHostedService(); @@ -48,12 +55,20 @@ public static IServiceCollection AddCrispyBackgroundJobs(this IServiceCollection /// Register a typed job handler where handler is resolved from DI and payload is serialized as JSON. /// handlerName is the logical identifier used when enqueueing jobs. /// - public static IServiceCollection AddJobHandler(this IServiceCollection services, string handlerName) + public static IServiceCollection AddJobHandler( + this IServiceCollection services, + string handlerName + ) where THandler : class, IBackgroundJobHandler { services.AddScoped(); // The registry registration happens at runtime; to ensure the registry has the entry we add a post-processor - services.AddSingleton(sp => new JobHandlerStartupFilter(sp, handlerName, typeof(THandler), typeof(TData))); + services.AddSingleton(sp => new JobHandlerStartupFilter( + sp, + handlerName, + typeof(THandler), + typeof(TData) + )); return services; } @@ -65,17 +80,30 @@ private class JobHandlerStartupFilter : Microsoft.AspNetCore.Hosting.IStartupFil private readonly Type _handlerType; private readonly Type _dataType; - public JobHandlerStartupFilter(IServiceProvider sp, string handlerName, Type handlerType, Type dataType) + public JobHandlerStartupFilter( + IServiceProvider sp, + string handlerName, + Type handlerType, + Type dataType + ) { - _sp = sp; _handlerName = handlerName; _handlerType = handlerType; _dataType = dataType; + _sp = sp; + _handlerName = handlerName; + _handlerType = handlerType; + _dataType = dataType; } - public Action Configure(Action next) + public Action Configure( + Action next + ) { return app => { var registry = _sp.GetRequiredService(); - var registerMethod = registry.GetType().GetMethod("Register")?.MakeGenericMethod(_handlerType, _dataType); + var registerMethod = registry + .GetType() + .GetMethod("Register") + ?.MakeGenericMethod(_handlerType, _dataType); // If Register(string) exists we call via reflection if (registerMethod != null) { diff --git a/Src/CrispyWaffle.BackgroundJobs/Monitoring/JobMetrics.cs b/Src/CrispyWaffle.BackgroundJobs/Monitoring/JobMetrics.cs index 58241216..726db9f5 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Monitoring/JobMetrics.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Monitoring/JobMetrics.cs @@ -13,7 +13,7 @@ public void Increment(string key) public long Get(string key) => _counters.TryGetValue(key, out var v) ? v : 0; - public System.Collections.Generic.IDictionary Snapshot() => new System.Collections.Generic.Dictionary(_counters); + public System.Collections.Generic.IDictionary Snapshot() => + new System.Collections.Generic.Dictionary(_counters); } } - diff --git a/Src/CrispyWaffle.BackgroundJobs/Monitoring/MonitoringMiddleware.cs b/Src/CrispyWaffle.BackgroundJobs/Monitoring/MonitoringMiddleware.cs index e4cd9c19..c9920c00 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Monitoring/MonitoringMiddleware.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Monitoring/MonitoringMiddleware.cs @@ -26,4 +26,4 @@ public async Task InvokeAsync(HttpContext context) await _next(context); } } -} \ No newline at end of file +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Persistence/EfCoreJobStore.cs b/Src/CrispyWaffle.BackgroundJobs/Persistence/EfCoreJobStore.cs index 74a2c7c3..3b1e4816 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Persistence/EfCoreJobStore.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Persistence/EfCoreJobStore.cs @@ -25,18 +25,22 @@ public async Task SaveAsync(JobEntity job, CancellationToken cancellationToken = var now = DateTimeOffset.UtcNow; // This uses a simple approach: select candidate IDs then try to update one row in a transaction; this reduces race conditions. - var candidate = await _db.Jobs - .Where(j => j.Status == JobStatus.Pending && (j.ScheduledAt == null || j.ScheduledAt <= now)) + var candidate = await _db + .Jobs.Where(j => + j.Status == JobStatus.Pending && (j.ScheduledAt == null || j.ScheduledAt <= now) + ) .OrderBy(j => (int)j.Priority) .ThenBy(j => j.CreatedAt) .Select(j => j.Id) .FirstOrDefaultAsync(cancellationToken); - if (candidate == Guid.Empty) return null; + if (candidate == Guid.Empty) + return null; // Try to claim the job var job = await _db.Jobs.FirstOrDefaultAsync(j => j.Id == candidate, cancellationToken); - if (job == null) return null; + if (job == null) + return null; job.Status = JobStatus.Processing; job.UpdatedAt = DateTimeOffset.UtcNow; @@ -53,29 +57,44 @@ public async Task SaveAsync(JobEntity job, CancellationToken cancellationToken = } } - public async Task MarkCompletedAsync(Guid jobId, CancellationToken cancellationToken = default) + public async Task MarkCompletedAsync( + Guid jobId, + CancellationToken cancellationToken = default + ) { var job = await _db.Jobs.FirstOrDefaultAsync(j => j.Id == jobId, cancellationToken); - if (job == null) return; + if (job == null) + return; job.Status = JobStatus.Completed; job.UpdatedAt = DateTimeOffset.UtcNow; await _db.SaveChangesAsync(cancellationToken); } - public async Task MarkFailedAsync(Guid jobId, string error, CancellationToken cancellationToken = default) + public async Task MarkFailedAsync( + Guid jobId, + string error, + CancellationToken cancellationToken = default + ) { var job = await _db.Jobs.FirstOrDefaultAsync(j => j.Id == jobId, cancellationToken); - if (job == null) return; + if (job == null) + return; job.Status = JobStatus.Failed; job.LastError = error; job.UpdatedAt = DateTimeOffset.UtcNow; await _db.SaveChangesAsync(cancellationToken); } - public async Task MarkRetryAsync(Guid jobId, DateTimeOffset? nextAttemptAt, int attemptCount, CancellationToken cancellationToken = default) + public async Task MarkRetryAsync( + Guid jobId, + DateTimeOffset? nextAttemptAt, + int attemptCount, + CancellationToken cancellationToken = default + ) { var job = await _db.Jobs.FirstOrDefaultAsync(j => j.Id == jobId, cancellationToken); - if (job == null) return; + if (job == null) + return; job.Attempt = attemptCount; job.Status = JobStatus.Pending; job.ScheduledAt = nextAttemptAt; @@ -83,4 +102,4 @@ public async Task MarkRetryAsync(Guid jobId, DateTimeOffset? nextAttemptAt, int await _db.SaveChangesAsync(cancellationToken); } } -} \ No newline at end of file +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Persistence/InMemoryJobStore.cs b/Src/CrispyWaffle.BackgroundJobs/Persistence/InMemoryJobStore.cs index c11a8d4e..ca4bac47 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Persistence/InMemoryJobStore.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Persistence/InMemoryJobStore.cs @@ -17,8 +17,11 @@ public Task SaveAsync(JobEntity job, CancellationToken cancellationToken = defau public Task FetchNextAsync(CancellationToken cancellationToken = default) { var now = DateTimeOffset.UtcNow; - var next = _storage.Values - .Where(j => j.Status == JobStatus.Pending && (!j.ScheduledAt.HasValue || j.ScheduledAt.Value <= now)) + var next = _storage + .Values.Where(j => + j.Status == JobStatus.Pending + && (!j.ScheduledAt.HasValue || j.ScheduledAt.Value <= now) + ) .OrderBy(j => (int)j.Priority) .ThenBy(j => j.CreatedAt) .FirstOrDefault(); @@ -46,7 +49,11 @@ public Task MarkCompletedAsync(Guid jobId, CancellationToken cancellationToken = return Task.CompletedTask; } - public Task MarkFailedAsync(Guid jobId, string error, CancellationToken cancellationToken = default) + public Task MarkFailedAsync( + Guid jobId, + string error, + CancellationToken cancellationToken = default + ) { if (_storage.TryGetValue(jobId, out var job)) { @@ -58,7 +65,12 @@ public Task MarkFailedAsync(Guid jobId, string error, CancellationToken cancella return Task.CompletedTask; } - public Task MarkRetryAsync(Guid jobId, DateTimeOffset? nextAttemptAt, int attemptCount, CancellationToken cancellationToken = default) + public Task MarkRetryAsync( + Guid jobId, + DateTimeOffset? nextAttemptAt, + int attemptCount, + CancellationToken cancellationToken = default + ) { if (_storage.TryGetValue(jobId, out var job)) { diff --git a/Src/CrispyWaffle.BackgroundJobs/Persistence/JobDbContext.cs b/Src/CrispyWaffle.BackgroundJobs/Persistence/JobDbContext.cs index ee719ad4..a81b112b 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Persistence/JobDbContext.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Persistence/JobDbContext.cs @@ -6,7 +6,8 @@ namespace CrispyWaffle.BackgroundJobs.Persistence { public class JobDbContext : DbContext { - public JobDbContext(DbContextOptions options) : base(options) { } + public JobDbContext(DbContextOptions options) + : base(options) { } public DbSet Jobs { get; set; } = null!; @@ -28,4 +29,4 @@ protected override void OnModelCreating(ModelBuilder modelBuilder) }); } } -} \ No newline at end of file +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Scheduling/CronJobScheduler.cs b/Src/CrispyWaffle.BackgroundJobs/Scheduling/CronJobScheduler.cs index b8acaa15..828f16e9 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Scheduling/CronJobScheduler.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Scheduling/CronJobScheduler.cs @@ -37,4 +37,4 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) protected abstract Task ExecuteOnce(CancellationToken stoppingToken); } -} \ No newline at end of file +} diff --git a/Src/CrispyWaffle.BackgroundJobs/Scheduling/DelayedJobScheduler.cs b/Src/CrispyWaffle.BackgroundJobs/Scheduling/DelayedJobScheduler.cs index 02194a82..54201717 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Scheduling/DelayedJobScheduler.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Scheduling/DelayedJobScheduler.cs @@ -16,16 +16,27 @@ public DelayedJobScheduler(IServiceProvider provider) _provider = provider; } - public Task EnqueueAsync(string handlerName, object payload, int maxAttempts = 3, JobPriority priority = JobPriority.Normal) + public Task EnqueueAsync( + string handlerName, + object payload, + int maxAttempts = 3, + JobPriority priority = JobPriority.Normal + ) { var dispatcher = _provider.GetRequiredService(); return dispatcher.EnqueueAsync(handlerName, payload, maxAttempts, priority); } - public Task ScheduleAsync(string handlerName, object payload, TimeSpan delay, int maxAttempts = 3, JobPriority priority = JobPriority.Normal) + public Task ScheduleAsync( + string handlerName, + object payload, + TimeSpan delay, + int maxAttempts = 3, + JobPriority priority = JobPriority.Normal + ) { var dispatcher = _provider.GetRequiredService(); return dispatcher.ScheduleAsync(handlerName, payload, delay, maxAttempts, priority); } } -} \ No newline at end of file +} From 54e5770efa99a25433f1f607a3938bc6fd713637 Mon Sep 17 00:00:00 2001 From: Guilherme Branco Stracini Date: Sat, 4 Jul 2026 22:58:39 +0100 Subject: [PATCH 09/10] refactor: extract methods for job fetching and processing Refactor `BackgroundWorker` by extracting methods for fetching and processing jobs, improving code readability and maintenance. This change introduces the following methods: `FetchNextJobAsync`, `RescheduleIfNotDueAsync`, and `ProcessJobWithScopeAsync`. In `HttpClientWrapper`, extracted methods to configure and apply HTTP client settings, enhancing clarity. New helper methods include `ConfigureBaseAddress`, `ConfigureTimeout`, `ApplyDefaultHeaders`, `EnsureAcceptHeader`, and `MergeRequestHeaders`. These changes promote better code organization and ease of understanding while maintaining existing functionality. --- .../Core/BackgroundWorker.cs | 322 ++++++++------- .../HttpClientWrapper.cs | 365 +++++++++++------- 2 files changed, 397 insertions(+), 290 deletions(-) diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs b/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs index d0a53871..533f7609 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs @@ -1,4 +1,4 @@ -using System.Text.Json; +using System.Text.Json; using CrispyWaffle.BackgroundJobs.Abstractions; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; @@ -53,90 +53,15 @@ private async Task WorkerLoopAsync(int workerId, CancellationToken stoppingToken { try { - JobEntity? job = null; - - if (_store != null) - { - job = await _store.FetchNextAsync(stoppingToken); - if (job == null) - { - await Task.Delay(_options.PollingInterval, stoppingToken); - continue; - } - } - else if (_queue != null) - { - job = await _queue.DequeueAsync(stoppingToken); - if (job == null) - { - await Task.Delay(TimeSpan.FromMilliseconds(200), stoppingToken); - continue; - } - } - else - { - _logger.LogWarning("No job queue nor store configured. Worker sleeping."); - await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken); - continue; - } - + var job = await FetchNextJobAsync(stoppingToken); if (job == null) continue; - // If job has ScheduledAt in future, re-enqueue / re-schedule. - if (job.ScheduledAt.HasValue && job.ScheduledAt.Value > DateTimeOffset.UtcNow) - { - // For persistence, mark back to pending with same ScheduledAt; for in-memory, just re-enqueue. - if (_store != null) - { - // mark as Pending again (store.FetchNextAsync should have returned only due jobs, this is defensive) - await _store.MarkRetryAsync( - job.Id, - job.ScheduledAt, - job.Attempt, - stoppingToken - ); - } - else - { - _queue!.Enqueue(job); - } - + if (await RescheduleIfNotDueAsync(job, stoppingToken)) continue; - } - - // Process job with scope so handlers can have scoped services. - using var scope = _provider.CreateScope(); - var scopedProvider = scope.ServiceProvider; - - var success = false; - try - { - success = await ProcessJobAsync(job, scopedProvider, stoppingToken); - } - catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) - { - // If shutting down, put the job back to pending (or leave DB in Processing if store does locking; here we mark retry) - if (_store != null) - { - var next = DateTimeOffset.UtcNow.AddSeconds(5); - await _store.MarkRetryAsync(job.Id, next, job.Attempt, stoppingToken); - } - else - { - _queue!.Enqueue(job); - } - break; // exit loop to allow graceful shutdown - } - - if (!success) - { - _logger.LogWarning( - "Job {JobId} failed after processing attempt {Attempt}", - job.Id, - job.Attempt - ); - } + + if (!await ProcessJobWithScopeAsync(job, stoppingToken)) + break; // shutting down } catch (Exception ex) { @@ -148,6 +73,92 @@ await _store.MarkRetryAsync( _logger.LogInformation("Worker {WorkerId} stopping", workerId); } + private async Task FetchNextJobAsync(CancellationToken stoppingToken) + { + if (_store != null) + { + var job = await _store.FetchNextAsync(stoppingToken); + if (job == null) + await Task.Delay(_options.PollingInterval, stoppingToken); + return job; + } + + if (_queue != null) + { + var job = await _queue.DequeueAsync(stoppingToken); + if (job == null) + await Task.Delay(TimeSpan.FromMilliseconds(200), stoppingToken); + return job; + } + + _logger.LogWarning("No job queue nor store configured. Worker sleeping."); + await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken); + return null; + } + + private async Task RescheduleIfNotDueAsync( + JobEntity job, + CancellationToken stoppingToken + ) + { + if (!job.ScheduledAt.HasValue || job.ScheduledAt.Value <= DateTimeOffset.UtcNow) + return false; + + // For persistence, mark back to pending with same ScheduledAt; for in-memory, just re-enqueue. + if (_store != null) + { + // mark as Pending again (store.FetchNextAsync should have returned only due jobs, this is defensive) + await _store.MarkRetryAsync(job.Id, job.ScheduledAt, job.Attempt, stoppingToken); + } + else + { + _queue!.Enqueue(job); + } + + return true; + } + + private async Task ProcessJobWithScopeAsync( + JobEntity job, + CancellationToken stoppingToken + ) + { + // Process job with scope so handlers can have scoped services. + using var scope = _provider.CreateScope(); + var scopedProvider = scope.ServiceProvider; + + bool success; + try + { + success = await ProcessJobAsync(job, scopedProvider, stoppingToken); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + // If shutting down, put the job back to pending (or leave DB in Processing if store does locking; here we mark retry) + if (_store != null) + { + var next = DateTimeOffset.UtcNow.AddSeconds(5); + await _store.MarkRetryAsync(job.Id, next, job.Attempt, stoppingToken); + } + else + { + _queue!.Enqueue(job); + } + return false; // exit loop to allow graceful shutdown + } + + if (!success) + { + _logger.LogWarning( + "Job {JobId} failed after processing attempt {Attempt}", + job.Id, + job.Attempt + ); + } + + return true; + } + private async Task ProcessJobAsync( JobEntity job, IServiceProvider scopedProvider, @@ -158,28 +169,26 @@ CancellationToken cancellationToken { if (!_registry.TryGet(job.HandlerName, out var handlerType, out var dataType)) { - var msg = $"No handler registered for '{job.HandlerName}'"; - _logger.LogError(msg); - if (_store != null) - await _store.MarkFailedAsync(job.Id, msg, cancellationToken); + await FailJobAsync( + job, + $"No handler registered for '{job.HandlerName}'", + cancellationToken + ); return false; } // Deserialize payload into dataType - object? data = null; - if (!string.IsNullOrWhiteSpace(job.Payload)) - { - data = JsonSerializer.Deserialize(job.Payload, dataType ?? typeof(object)); - } + var data = DeserializePayload(job, dataType); // Resolve handler instance var handler = scopedProvider.GetService(handlerType!); if (handler == null) { - var msg = $"Handler type {handlerType} not resolved from DI"; - _logger.LogError(msg); - if (_store != null) - await _store.MarkFailedAsync(job.Id, msg, cancellationToken); + await FailJobAsync( + job, + $"Handler type {handlerType} not resolved from DI", + cancellationToken + ); return false; } @@ -193,62 +202,53 @@ CancellationToken cancellationToken return true; } - // handle retry - job.Attempt++; - var shouldRetry = result.Retry && job.Attempt < job.MaxAttempt; - if (shouldRetry) - { - var backoffSeconds = ComputeBackoffSeconds(job.Attempt); - var nextAttempt = DateTimeOffset.UtcNow.AddSeconds(backoffSeconds); - _logger.LogInformation( - "Scheduling retry for job {JobId} after {Delay}s (attempt {Attempt}/{MaxAttempt})", - job.Id, - backoffSeconds, - job.Attempt, - job.MaxAttempt - ); - - if (_store != null) - { - await _store.MarkRetryAsync( - job.Id, - nextAttempt, - job.Attempt, - cancellationToken - ); - } - else - { - job.ScheduledAt = nextAttempt; - _queue!.Enqueue(job); - } - } - else - { - // exhaust - var err = result.ErrorMessage ?? "Job failed"; - if (_store != null) - await _store.MarkFailedAsync(job.Id, err, cancellationToken); - else - _logger.LogError( - "Job {JobId} failed and will not be retried: {Error}", - job.Id, - err - ); - } - + await HandleJobFailureAsync(job, result, cancellationToken); return false; } catch (Exception ex) { _logger.LogError(ex, "Exception when processing job {JobId}", job.Id); + await ScheduleRetryAsync(job, cancellationToken); + return false; + } + } + + private async Task FailJobAsync(JobEntity job, string message, CancellationToken cancellationToken) + { + _logger.LogError(message); + if (_store != null) + await _store.MarkFailedAsync(job.Id, message, cancellationToken); + } + + private static object? DeserializePayload(JobEntity job, Type? dataType) + { + if (string.IsNullOrWhiteSpace(job.Payload)) + return null; + return JsonSerializer.Deserialize(job.Payload, dataType ?? typeof(object)); + } + + private async Task HandleJobFailureAsync( + JobEntity job, + JobResult result, + CancellationToken cancellationToken + ) + { + job.Attempt++; + var shouldRetry = result.Retry && job.Attempt < job.MaxAttempt; + if (shouldRetry) + { + var backoffSeconds = ComputeBackoffSeconds(job.Attempt); + var nextAttempt = DateTimeOffset.UtcNow.AddSeconds(backoffSeconds); + _logger.LogInformation( + "Scheduling retry for job {JobId} after {Delay}s (attempt {Attempt}/{MaxAttempt})", + job.Id, + backoffSeconds, + job.Attempt, + job.MaxAttempt + ); + if (_store != null) { - // schedule retry with backoff - job.Attempt++; - var nextAttempt = DateTimeOffset.UtcNow.AddSeconds( - ComputeBackoffSeconds(job.Attempt) - ); await _store.MarkRetryAsync( job.Id, nextAttempt, @@ -258,14 +258,40 @@ await _store.MarkRetryAsync( } else { - job.Attempt++; - job.ScheduledAt = DateTimeOffset.UtcNow.AddSeconds( - ComputeBackoffSeconds(job.Attempt) - ); + job.ScheduledAt = nextAttempt; _queue!.Enqueue(job); } + } + else + { + // exhaust + var err = result.ErrorMessage ?? "Job failed"; + if (_store != null) + await _store.MarkFailedAsync(job.Id, err, cancellationToken); + else + _logger.LogError( + "Job {JobId} failed and will not be retried: {Error}", + job.Id, + err + ); + } + } - return false; + private async Task ScheduleRetryAsync(JobEntity job, CancellationToken cancellationToken) + { + job.Attempt++; + var nextAttempt = DateTimeOffset.UtcNow.AddSeconds( + ComputeBackoffSeconds(job.Attempt) + ); + + if (_store != null) + { + await _store.MarkRetryAsync(job.Id, nextAttempt, job.Attempt, cancellationToken); + } + else + { + job.ScheduledAt = nextAttempt; + _queue!.Enqueue(job); } } diff --git a/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs b/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs index 2ad342d0..8d014493 100644 --- a/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs +++ b/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs @@ -45,29 +45,45 @@ private System.Net.Http.HttpClient CreateClient(HttpRequestOptions? perRequest) ? _httpClientFactory.CreateClient() : _httpClientFactory.CreateClient(name); + ConfigureBaseAddress(client, perRequest); + ConfigureTimeout(client, perRequest); + ApplyDefaultHeaders(client, perRequest); + EnsureAcceptHeader(client); + + return client; + } + + private void ConfigureBaseAddress( + System.Net.Http.HttpClient client, + HttpRequestOptions? perRequest + ) + { var baseAddr = perRequest?.BaseAddress ?? _defaults.BaseAddress; if (!string.IsNullOrEmpty(baseAddr) && client.BaseAddress == null) { client.BaseAddress = new Uri(baseAddr); } + } + private void ConfigureTimeout( + System.Net.Http.HttpClient client, + HttpRequestOptions? perRequest + ) + { var timeoutSeconds = perRequest?.TimeoutSeconds ?? _defaults.TimeoutSeconds; client.Timeout = TimeSpan.FromSeconds(Math.Max(1, timeoutSeconds)); + } - // Merge default headers (global) and per-request headers - var headers = new Dictionary(); - if (_defaults.DefaultRequestHeaders != null) - { - foreach (var kv in _defaults.DefaultRequestHeaders) - headers[kv.Key] = kv.Value; - } - if (perRequest?.DefaultRequestHeaders != null) - { - foreach (var kv in perRequest.DefaultRequestHeaders) - headers[kv.Key] = kv.Value; - } + private void ApplyDefaultHeaders( + System.Net.Http.HttpClient client, + HttpRequestOptions? perRequest + ) + { + var headers = MergeRequestHeaders( + _defaults.DefaultRequestHeaders, + perRequest?.DefaultRequestHeaders + ); - // Set headers (be careful to not duplicate) foreach (var kv in headers) { if (!client.DefaultRequestHeaders.Contains(kv.Key)) @@ -82,16 +98,35 @@ private System.Net.Http.HttpClient CreateClient(HttpRequestOptions? perRequest) } } } + } - // Ensure Accept header + private static void EnsureAcceptHeader(System.Net.Http.HttpClient client) + { if (client.DefaultRequestHeaders.Accept.Count == 0) { client.DefaultRequestHeaders.Accept.Add( new System.Net.Http.Headers.MediaTypeWithQualityHeaderValue("application/json") ); } + } - return client; + private static Dictionary MergeRequestHeaders( + IDictionary? defaults, + IDictionary? perRequest + ) + { + var headers = new Dictionary(); + if (defaults != null) + { + foreach (var kv in defaults) + headers[kv.Key] = kv.Value; + } + if (perRequest != null) + { + foreach (var kv in perRequest) + headers[kv.Key] = kv.Value; + } + return headers; } #region Public Async Methods @@ -168,7 +203,52 @@ CancellationToken cancellationToken var sw = Stopwatch.StartNew(); // The action will create a NEW HttpRequestMessage each time it's invoked (important for retries). - Func> action = async ct => + var action = BuildRequestAction(client, method, url, body, effectiveOptions, serializer); + + // Execute the action with retry policy. The policy is expected to return the final response (or null + exception). + var (resp, ex) = await RetryPolicy + .ExecuteAsync( + action, + effectiveOptions.RetryCount, + effectiveOptions.RetryBaseDelayMs, + effectiveOptions.MaxRetryDelayMs, + cancellationToken + ) + .ConfigureAwait(false); + + sw.Stop(); + + // If no response after retries + if (resp == null) + { + return BuildNoResponseResult(effectiveOptions, ex, sw.Elapsed); + } + + // IMPORTANT: dispose HttpResponseMessage after we've finished reading headers/body. + using (resp) + { + return await BuildResponseResultAsync( + resp, + method, + url, + serializer, + effectiveOptions, + sw.Elapsed + ) + .ConfigureAwait(false); + } + } + + private Func> BuildRequestAction( + System.Net.Http.HttpClient client, + HttpMethod method, + string url, + object? body, + HttpRequestOptions effectiveOptions, + IJsonSerializer serializer + ) + { + return async ct => { // Create a new request per attempt - ensures headers/content are fresh for retries. using var req = new HttpRequestMessage(method, url); @@ -200,157 +280,158 @@ CancellationToken cancellationToken .SendAsync(req, HttpCompletionOption.ResponseHeadersRead, ct) .ConfigureAwait(false); }; + } - // Execute the action with retry policy. The policy is expected to return the final response (or null + exception). - var (resp, ex) = await RetryPolicy - .ExecuteAsync( - action, - effectiveOptions.RetryCount, - effectiveOptions.RetryBaseDelayMs, - effectiveOptions.MaxRetryDelayMs, - cancellationToken - ) - .ConfigureAwait(false); + private static HttpResponse BuildNoResponseResult( + HttpRequestOptions effectiveOptions, + Exception? ex, + TimeSpan elapsed + ) + { + if (effectiveOptions.ThrowOnError) + { + throw new HttpClientException("HTTP request failed after retries", null, null, ex); + } - sw.Stop(); + var errors = new List { ex?.Message ?? "Unknown error" }; + return HttpResponse.Failure(null, null, errors, null, elapsed); + } - // If no response after retries - if (resp == null) + private async Task> BuildResponseResultAsync( + HttpResponseMessage resp, + HttpMethod method, + string url, + IJsonSerializer serializer, + HttpRequestOptions effectiveOptions, + TimeSpan elapsed + ) + { + string? raw; + try + { + // If there is content, read it as string. Keep it simple and synchronous from serializer's perspective. + raw = + resp.Content != null + ? await resp.Content.ReadAsStringAsync().ConfigureAwait(false) + : null; + } + catch (Exception readEx) { + // Failed to read content. + _logger?.LogError( + readEx, + "Failed to read response content for {Method} {Url}", + method, + url + ); if (effectiveOptions.ThrowOnError) { throw new HttpClientException( - "HTTP request failed after retries", - null, + "Failed to read response content", + resp.StatusCode, null, - ex + readEx ); } - var errors = new List { ex?.Message ?? "Unknown error" }; - return HttpResponse.Failure(null, null, errors, null, sw.Elapsed); + var readErrors = new List { readEx.Message }; + return HttpResponse.Failure( + resp.StatusCode, + null, + readErrors, + resp.Headers.ToDictionary(h => h.Key, h => h.Value.AsEnumerable()), + elapsed + ); } - // IMPORTANT: dispose HttpResponseMessage after we've finished reading headers/body. - using (resp) + // Gather headers (including content headers). + var headers = resp.Headers.ToDictionary(h => h.Key, h => h.Value.AsEnumerable()); + if (resp.Content?.Headers != null) { - string? raw = null; - try + foreach (var h in resp.Content.Headers) + headers[h.Key] = h.Value.AsEnumerable(); + } + + return resp.IsSuccessStatusCode + ? BuildSuccessResult(resp, raw, method, url, serializer, effectiveOptions, headers, elapsed) + : BuildFailureResult(resp, raw, headers, effectiveOptions, elapsed); + } + + private HttpResponse BuildSuccessResult( + HttpResponseMessage resp, + string? raw, + HttpMethod method, + string url, + IJsonSerializer serializer, + HttpRequestOptions effectiveOptions, + IDictionary> headers, + TimeSpan elapsed + ) + { + T? data = default; + try + { + // If raw body is present, try to deserialize; otherwise default(T). + if (!string.IsNullOrWhiteSpace(raw)) { - // If there is content, read it as string. Keep it simple and synchronous from serializer's perspective. - if (resp.Content != null) - { - raw = await resp.Content.ReadAsStringAsync().ConfigureAwait(false); - } + data = serializer.Deserialize(raw); } - catch (Exception readEx) + } + catch (Exception deserEx) + { + // Deserialization failed - log and either throw or return failure depending on options. + _logger?.LogError( + deserEx, + "Failed to deserialize response body to {Type} for {Method} {Url}", + typeof(T).FullName, + method, + url + ); + if (effectiveOptions.ThrowOnError) { - // Failed to read content. - _logger?.LogError( - readEx, - "Failed to read response content for {Method} {Url}", - method, - url - ); - if (effectiveOptions.ThrowOnError) - { - throw new HttpClientException( - "Failed to read response content", - resp.StatusCode, - null, - readEx - ); - } - - var readErrors = new List { readEx.Message }; - return HttpResponse.Failure( + throw new HttpClientException( + $"Failed to deserialize response to {typeof(T)}", resp.StatusCode, - null, - readErrors, - resp.Headers.ToDictionary(h => h.Key, h => h.Value.AsEnumerable()), - sw.Elapsed + raw, + deserEx ); } - // Gather headers (including content headers). - var headers = resp.Headers.ToDictionary(h => h.Key, h => h.Value.AsEnumerable()); - if (resp.Content?.Headers != null) - { - foreach (var h in resp.Content.Headers) - headers[h.Key] = h.Value.AsEnumerable(); - } - - // Successful HTTP status - if (resp.IsSuccessStatusCode) - { - T? data = default; - try - { - // If raw body is present, try to deserialize; otherwise default(T). - if (!string.IsNullOrWhiteSpace(raw)) - { - data = serializer.Deserialize(raw); - } - } - catch (Exception deserEx) - { - // Deserialization failed - log and either throw or return failure depending on options. - _logger?.LogError( - deserEx, - "Failed to deserialize response body to {Type} for {Method} {Url}", - typeof(T).FullName, - method, - url - ); - if (effectiveOptions.ThrowOnError) - { - throw new HttpClientException( - $"Failed to deserialize response to {typeof(T)}", - resp.StatusCode, - raw, - deserEx - ); - } - - return HttpResponse.Failure( - resp.StatusCode, - raw, - new List { "DeserializationFailed: " + deserEx.Message }, - headers, - sw.Elapsed - ); - } + return HttpResponse.Failure( + resp.StatusCode, + raw, + new List { "DeserializationFailed: " + deserEx.Message }, + headers, + elapsed + ); + } - return HttpResponse.Success(data, resp.StatusCode, raw, headers, sw.Elapsed); - } - else - { - // Non-success HTTP status - var errors = new List - { - $"HTTP {(int)resp.StatusCode} {resp.ReasonPhrase}", - }; - if (!string.IsNullOrWhiteSpace(raw)) - errors.Add(raw); + return HttpResponse.Success(data, resp.StatusCode, raw, headers, elapsed); + } - if (effectiveOptions.ThrowOnError) - { - throw new HttpClientException( - $"HTTP request failed with {(int)resp.StatusCode}", - resp.StatusCode, - raw - ); - } + private static HttpResponse BuildFailureResult( + HttpResponseMessage resp, + string? raw, + IDictionary> headers, + HttpRequestOptions effectiveOptions, + TimeSpan elapsed + ) + { + // Non-success HTTP status + var errors = new List { $"HTTP {(int)resp.StatusCode} {resp.ReasonPhrase}" }; + if (!string.IsNullOrWhiteSpace(raw)) + errors.Add(raw); - return HttpResponse.Failure( - resp.StatusCode, - raw, - errors, - headers, - sw.Elapsed - ); - } + if (effectiveOptions.ThrowOnError) + { + throw new HttpClientException( + $"HTTP request failed with {(int)resp.StatusCode}", + resp.StatusCode, + raw + ); } + + return HttpResponse.Failure(resp.StatusCode, raw, errors, headers, elapsed); } #endregion From f4af0a245e13345b0174fc94013bb0f83b6faf96 Mon Sep 17 00:00:00 2001 From: Guilherme Branco Stracini Date: Sat, 4 Jul 2026 23:01:45 +0100 Subject: [PATCH 10/10] refactor: improve code readability with formatting adjustments Improve code readability by reformatting method signatures and block structures to follow consistent line breaking and indentation practices. This affects the BackgroundWorker and HttpClientWrapper classes, where method headers are now more uniformly formatted, enhancing maintainability. These changes were done to ensure that the code follows better styling conventions and makes the codebase more consistent and easier to read. --- .../Core/BackgroundWorker.cs | 10 ++++++---- .../HttpClientWrapper.cs | 20 +++++++++++++++++-- 2 files changed, 24 insertions(+), 6 deletions(-) diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs b/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs index 533f7609..aeb30985 100644 --- a/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs +++ b/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs @@ -213,7 +213,11 @@ await FailJobAsync( } } - private async Task FailJobAsync(JobEntity job, string message, CancellationToken cancellationToken) + private async Task FailJobAsync( + JobEntity job, + string message, + CancellationToken cancellationToken + ) { _logger.LogError(message); if (_store != null) @@ -280,9 +284,7 @@ await _store.MarkRetryAsync( private async Task ScheduleRetryAsync(JobEntity job, CancellationToken cancellationToken) { job.Attempt++; - var nextAttempt = DateTimeOffset.UtcNow.AddSeconds( - ComputeBackoffSeconds(job.Attempt) - ); + var nextAttempt = DateTimeOffset.UtcNow.AddSeconds(ComputeBackoffSeconds(job.Attempt)); if (_store != null) { diff --git a/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs b/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs index 8d014493..46a28511 100644 --- a/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs +++ b/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs @@ -203,7 +203,14 @@ CancellationToken cancellationToken var sw = Stopwatch.StartNew(); // The action will create a NEW HttpRequestMessage each time it's invoked (important for retries). - var action = BuildRequestAction(client, method, url, body, effectiveOptions, serializer); + var action = BuildRequestAction( + client, + method, + url, + body, + effectiveOptions, + serializer + ); // Execute the action with retry policy. The policy is expected to return the final response (or null + exception). var (resp, ex) = await RetryPolicy @@ -353,7 +360,16 @@ TimeSpan elapsed } return resp.IsSuccessStatusCode - ? BuildSuccessResult(resp, raw, method, url, serializer, effectiveOptions, headers, elapsed) + ? BuildSuccessResult( + resp, + raw, + method, + url, + serializer, + effectiveOptions, + headers, + elapsed + ) : BuildFailureResult(resp, raw, headers, effectiveOptions, elapsed); }