diff --git a/CHANGELOG.md b/CHANGELOG.md index 9bc3fe6..46cf604 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,18 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [2.1.0] - 2026-07-30 + +### Added +- **HA work leases via Dapr state:** `IDaprWorkLeaseStore` / `DaprWorkLeaseStore` (acquire, renew, release, get) using ETag concurrency — broker-agnostic (NATS KV / Postgres / other state Component). +- **State ETag API:** `GetStateAndETagAsync` and `TrySaveStateAsync` on `IDaprStateStoreService`. +- **Runtime instance id:** `IDaprRuntimeInstanceId` / `DaprRuntimeInstanceIdProvider` (`POD_NAME` in Kubernetes). +- **Pub/sub worker helpers:** `IDaprPubSubWorkHandler`, `DaprPubSubAcceptOutcome` / `DaprPubSubAcceptResult`, `DaprPubSubAck` (HTTP ACK/NAK mapping). +- DI: `RegisterWorkLeases()` registers state store + lease store + instance id. + +### Changed +- Package version **2.1.0**. + ## [2.0.1] - 2026-06-28 ### Changed @@ -25,27 +37,3 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Removed - Legacy root-level release scripts (`Release-NuGetPackage.*`) in favor of the `utils/Release-NuGetPackage/` flow. - - diff --git a/README.md b/README.md index 82a608f..90ed771 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,8 @@ # MaksIT.Dapr -![Line Coverage](https://img.shields.io/badge/Line%20Coverage-62.1%25-green) -![Branch Coverage](https://img.shields.io/badge/Branch%20Coverage-50%25-yellowgreen) -![Method Coverage](https://img.shields.io/badge/Method%20Coverage-60%25-green) +![Line Coverage](https://img.shields.io/badge/Line%20Coverage-41.3%25-yellowgreen) +![Branch Coverage](https://img.shields.io/badge/Branch%20Coverage-25%25-yellow) +![Method Coverage](https://img.shields.io/badge/Method%20Coverage-53.1%25-yellowgreen) This repository hosts the `maksit-dapr` project, which utilizes [Dapr](https://dapr.io/) (Distributed Application Runtime) to facilitate building and managing microservices with ease. The project focuses on implementing a robust, scalable solution leveraging Dapr's building blocks and abstractions. @@ -67,6 +67,8 @@ var builder = WebApplication.CreateBuilder(args); // Register Dapr services builder.Services.RegisterPublisher(); builder.Services.RegisterStateStore(); +// HA coordination (leases via Dapr state Component — e.g. persistent NATS KV) +builder.Services.RegisterWorkLeases(); var app = builder.Build(); @@ -166,6 +168,28 @@ public class MyService This setup enables your ASP.NET Core application to utilize Dapr's pub-sub, state management, and other building blocks with minimal boilerplate. +### HA work leases (multi-replica) + +Use Dapr **state** (not product DB lock tables) for bootstrap/sweep/holder coordination. The state Component is infrastructure (e.g. persistent NATS JetStream KV); application code stays broker-agnostic. + +```csharp +builder.Services.RegisterWorkLeases(); + +// inject IDaprWorkLeaseStore + IDaprRuntimeInstanceId +var acquired = await leases.TryAcquireAsync( + storeName: "maksit-cicd-state", + workKey: "bootstrap", + holderId: runtimeInstance.InstanceId, + ttl: TimeSpan.FromMinutes(5), + ct); + +if (acquired is { IsSuccess: true, Value: true }) { + try { /* exclusive work */ } + finally { await leases.ReleaseAsync("maksit-cicd-state", "bootstrap", runtimeInstance.InstanceId, CancellationToken.None); } +} +``` + +Pub/sub workers: implement `IDaprPubSubWorkHandler` and map with `DaprPubSubAck.ToActionResult` (2xx ACK, 503 Busy → redelivery). ## Contributing diff --git a/src/MaksIT.Dapr.Tests/DaprWorkLeaseStoreTests.cs b/src/MaksIT.Dapr.Tests/DaprWorkLeaseStoreTests.cs new file mode 100644 index 0000000..5dac00a --- /dev/null +++ b/src/MaksIT.Dapr.Tests/DaprWorkLeaseStoreTests.cs @@ -0,0 +1,82 @@ +using Dapr.Client; +using MaksIT.Dapr.PubSub; +using MaksIT.Dapr.Services; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Mvc; +using Microsoft.Extensions.Logging; +using Moq; + + +namespace MaksIT.Dapr.Tests; + +public class DaprWorkLeaseStoreTests { + [Fact] + public async Task TryAcquireAsync_Succeeds_WhenKeyMissing() { + var state = new Mock(); + state + .Setup(s => s.GetStateAndETagAsync("store", "work", It.IsAny())) + .ReturnsAsync(MaksIT.Results.Result<(DaprWorkLease? Value, string? ETag)>.Ok((null, null))); + state + .Setup(s => s.TrySaveStateAsync("store", "work", It.IsAny(), null, It.IsAny())) + .ReturnsAsync(MaksIT.Results.Result.Ok(true)); + + var store = new DaprWorkLeaseStore(state.Object); + var result = await store.TryAcquireAsync("store", "work", "pod-a", TimeSpan.FromMinutes(1)); + + Assert.True(result.IsSuccess); + Assert.True(result.Value); + } + + [Fact] + public async Task TryAcquireAsync_Fails_WhenHeldByOtherAndNotExpired() { + var lease = new DaprWorkLease("pod-b", DateTimeOffset.UtcNow, DateTimeOffset.UtcNow.AddMinutes(5)); + var state = new Mock(); + state + .Setup(s => s.GetStateAndETagAsync("store", "work", It.IsAny())) + .ReturnsAsync(MaksIT.Results.Result<(DaprWorkLease? Value, string? ETag)>.Ok((lease, "1"))); + + var store = new DaprWorkLeaseStore(state.Object); + var result = await store.TryAcquireAsync("store", "work", "pod-a", TimeSpan.FromMinutes(1)); + + Assert.True(result.IsSuccess); + Assert.False(result.Value); + } +} + +public class DaprPubSubAckTests { + [Fact] + public void ToActionResult_Busy_Returns503() { + var result = DaprPubSubAck.ToActionResult(DaprPubSubAcceptResult.Busy("full")); + var objectResult = Assert.IsType(result); + Assert.Equal(StatusCodes.Status503ServiceUnavailable, objectResult.StatusCode); + } + + [Fact] + public void ToActionResult_Accepted_Returns200() { + var result = DaprPubSubAck.ToActionResult(DaprPubSubAcceptResult.Accepted()); + Assert.IsType(result); + } +} + +public class DaprStateStoreETagTests { + [Fact] + public async Task TrySaveStateAsync_ReturnsOkFalse_WhenClientReturnsFalse() { + var clientMock = new Mock(); + clientMock + .Setup(x => x.TrySaveStateAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny>(), + It.IsAny())) + .ReturnsAsync(false); + + var service = new DaprStateStoreService(Mock.Of>(), clientMock.Object); + var result = await service.TrySaveStateAsync("store", "key", "value", "etag"); + + Assert.True(result.IsSuccess); + Assert.False(result.Value); + } +} diff --git a/src/MaksIT.Dapr/Extensions/ServiceCollectionExtensions.cs b/src/MaksIT.Dapr/Extensions/ServiceCollectionExtensions.cs index 3bea7c8..eb6f05c 100644 --- a/src/MaksIT.Dapr/Extensions/ServiceCollectionExtensions.cs +++ b/src/MaksIT.Dapr/Extensions/ServiceCollectionExtensions.cs @@ -5,13 +5,14 @@ using Microsoft.Extensions.DependencyInjection; namespace MaksIT.Dapr.Extensions; public static class ServiceCollectionExtensions { - private static bool _isDaprClientRegistered = false; + private static bool _isDaprClientRegistered; private static void AddDaprClientOnce(this IServiceCollection services) { - if (!_isDaprClientRegistered) { - services.AddDaprClient(); - _isDaprClientRegistered = true; - } + if (_isDaprClientRegistered) + return; + + services.AddDaprClient(); + _isDaprClientRegistered = true; } public static void RegisterPublisher(this IServiceCollection services) { @@ -23,4 +24,13 @@ public static class ServiceCollectionExtensions { services.AddDaprClientOnce(); services.AddSingleton(); } -} \ No newline at end of file + + /// + /// Registers Dapr state store plus HA work-lease coordination and runtime instance id. + /// + public static void RegisterWorkLeases(this IServiceCollection services) { + services.RegisterStateStore(); + services.AddSingleton(); + services.AddSingleton(); + } +} diff --git a/src/MaksIT.Dapr/MaksIT.Dapr.csproj b/src/MaksIT.Dapr/MaksIT.Dapr.csproj index b77270e..2c55f6c 100644 --- a/src/MaksIT.Dapr/MaksIT.Dapr.csproj +++ b/src/MaksIT.Dapr/MaksIT.Dapr.csproj @@ -7,7 +7,7 @@ MaksIT.Dapr - 2.0.1 + 2.1.0 Maksym Sadovnychyy MAKS-IT MaksIT.Dapr diff --git a/src/MaksIT.Dapr/PubSub/DaprPubSubWork.cs b/src/MaksIT.Dapr/PubSub/DaprPubSubWork.cs new file mode 100644 index 0000000..f08cab1 --- /dev/null +++ b/src/MaksIT.Dapr/PubSub/DaprPubSubWork.cs @@ -0,0 +1,45 @@ +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Mvc; + + +namespace MaksIT.Dapr.PubSub; + +public enum DaprPubSubAcceptOutcome { + Accepted = 0, + AlreadyHandled = 1, + Busy = 2, + Rejected = 3, +} + +public sealed record DaprPubSubAcceptResult(DaprPubSubAcceptOutcome Outcome, string? Detail = null) { + public static DaprPubSubAcceptResult Accepted(string? detail = null) => new(DaprPubSubAcceptOutcome.Accepted, detail); + public static DaprPubSubAcceptResult AlreadyHandled(string? detail = null) => new(DaprPubSubAcceptOutcome.AlreadyHandled, detail); + public static DaprPubSubAcceptResult Busy(string? detail = null) => new(DaprPubSubAcceptOutcome.Busy, detail); + public static DaprPubSubAcceptResult Rejected(string? detail = null) => new(DaprPubSubAcceptOutcome.Rejected, detail); +} + +/// Product implements accept/claim; HTTP layer maps to Dapr ACK/NAK via . +public interface IDaprPubSubWorkHandler { + Task TryAcceptAsync(TMessage message, CancellationToken cancellationToken = default); +} + +/// Maps accept outcomes to HTTP status codes for Dapr pub/sub delivery. +public static class DaprPubSubAck { + public static IActionResult ToActionResult(DaprPubSubAcceptResult result) => + result.Outcome switch { + DaprPubSubAcceptOutcome.Accepted => new OkObjectResult(new { status = "accepted", detail = result.Detail }), + DaprPubSubAcceptOutcome.AlreadyHandled => new OkObjectResult(new { status = "alreadyHandled", detail = result.Detail }), + DaprPubSubAcceptOutcome.Busy => new ObjectResult(new { status = "busy", detail = result.Detail }) { StatusCode = StatusCodes.Status503ServiceUnavailable }, + DaprPubSubAcceptOutcome.Rejected => new BadRequestObjectResult(new { status = "rejected", detail = result.Detail }), + _ => new StatusCodeResult(StatusCodes.Status500InternalServerError), + }; + + public static Microsoft.AspNetCore.Http.IResult ToHttpResult(DaprPubSubAcceptResult result) => + result.Outcome switch { + DaprPubSubAcceptOutcome.Accepted => Microsoft.AspNetCore.Http.Results.Ok(new { status = "accepted", detail = result.Detail }), + DaprPubSubAcceptOutcome.AlreadyHandled => Microsoft.AspNetCore.Http.Results.Ok(new { status = "alreadyHandled", detail = result.Detail }), + DaprPubSubAcceptOutcome.Busy => Microsoft.AspNetCore.Http.Results.Json(new { status = "busy", detail = result.Detail }, statusCode: StatusCodes.Status503ServiceUnavailable), + DaprPubSubAcceptOutcome.Rejected => Microsoft.AspNetCore.Http.Results.BadRequest(new { status = "rejected", detail = result.Detail }), + _ => Microsoft.AspNetCore.Http.Results.StatusCode(StatusCodes.Status500InternalServerError), + }; +} diff --git a/src/MaksIT.Dapr/Services/DaprRuntimeInstanceIdProvider.cs b/src/MaksIT.Dapr/Services/DaprRuntimeInstanceIdProvider.cs new file mode 100644 index 0000000..9948b18 --- /dev/null +++ b/src/MaksIT.Dapr/Services/DaprRuntimeInstanceIdProvider.cs @@ -0,0 +1,26 @@ +namespace MaksIT.Dapr.Services; + +/// Stable id for this process/pod (lease holder). +public interface IDaprRuntimeInstanceId { + string InstanceId { get; } +} + +/// +/// Prefers POD_NAME in Kubernetes; otherwise host name + process id. +/// +public sealed class DaprRuntimeInstanceIdProvider : IDaprRuntimeInstanceId { + public string InstanceId { get; } = Build(); + + private static string Build() { + var logicalHost = + Environment.GetEnvironmentVariable("POD_NAME") + ?? Environment.GetEnvironmentVariable("HOSTNAME") + ?? Environment.GetEnvironmentVariable("COMPUTERNAME") + ?? Environment.MachineName; + + if (!string.IsNullOrEmpty(Environment.GetEnvironmentVariable("KUBERNETES_SERVICE_HOST"))) + return logicalHost; + + return $"{logicalHost}-{Environment.ProcessId}"; + } +} diff --git a/src/MaksIT.Dapr/Services/DaprStateStoreService.cs b/src/MaksIT.Dapr/Services/DaprStateStoreService.cs index 0db132b..12a881b 100644 --- a/src/MaksIT.Dapr/Services/DaprStateStoreService.cs +++ b/src/MaksIT.Dapr/Services/DaprStateStoreService.cs @@ -1,85 +1,93 @@ -using Microsoft.Extensions.Logging; - -using Dapr.Client; - -using MaksIT.Results; +using Dapr.Client; using MaksIT.Core.Extensions; +using MaksIT.Results; +using Microsoft.Extensions.Logging; + namespace MaksIT.Dapr.Services; + public interface IDaprStateStoreService { Task SetStateAsync(string storeName, string key, T value); Task> GetStateAsync(string storeName, string key); + Task> GetStateAndETagAsync(string storeName, string key, CancellationToken cancellationToken = default); + Task> TrySaveStateAsync(string storeName, string key, T value, string? etag, CancellationToken cancellationToken = default); Task DeleteStateAsync(string storeName, string key); } - public class DaprStateStoreService : IDaprStateStoreService { - private const string _errorMessage = "MaksIT.Dapr - Data provider error"; + private const string ErrorMessage = "MaksIT.Dapr - Data provider error"; private readonly DaprClient _client; private readonly ILogger _logger; - public DaprStateStoreService( - ILogger logger, - DaprClient client - ) { + public DaprStateStoreService(ILogger logger, DaprClient client) { _logger = logger; _client = client; } - /// - /// Saves a state to a Dapr state store - /// - /// - /// - /// - /// - /// + public async Task SetStateAsync(string storeName, string key, T value) { try { await _client.SaveStateAsync(storeName, key, value); return Result.Ok(); } catch (Exception ex) { - _logger.LogError(ex, _errorMessage); - return Result.InternalServerError(new[] {_errorMessage}.Concat(ex.ExtractMessages()).ToArray()); + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); } } - /// - /// Gets a state from a Dapr state store - /// - /// - /// - /// - /// public async Task> GetStateAsync(string storeName, string key) { try { var state = await _client.GetStateAsync(storeName, key); - if (state == null) + if (state is null) return Result.NotFound(default, $"State from the store {storeName} with the {key} not found."); return Result.Ok(state); } catch (Exception ex) { - _logger.LogError(ex, _errorMessage); - return Result.InternalServerError(default, new[] {_errorMessage}.Concat(ex.ExtractMessages()).ToArray()); + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(default, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + public async Task> GetStateAndETagAsync( + string storeName, + string key, + CancellationToken cancellationToken = default) { + try { + var (value, etag) = await _client.GetStateAndETagAsync(storeName, key, cancellationToken: cancellationToken); + return Result<(T? Value, string? ETag)>.Ok((value, etag)); + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result<(T? Value, string? ETag)>.InternalServerError(default, [ErrorMessage, .. ex.ExtractMessages()]); + } + } + + public async Task> TrySaveStateAsync( + string storeName, + string key, + T value, + string? etag, + CancellationToken cancellationToken = default) { + try { + var saved = await _client.TrySaveStateAsync(storeName, key, value, etag ?? string.Empty, cancellationToken: cancellationToken); + return Result.Ok(saved); + } + catch (Exception ex) { + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError(false, [ErrorMessage, .. ex.ExtractMessages()]); } } - /// - /// Deletes a state from a Dapr state store - /// - /// - /// - /// public async Task DeleteStateAsync(string storeName, string key) { try { await _client.DeleteStateAsync(storeName, key); return Result.Ok(); } catch (Exception ex) { - _logger.LogError(ex, _errorMessage); - return Result.InternalServerError([_errorMessage, .. ex.ExtractMessages()]); + _logger.LogError(ex, ErrorMessage); + return Result.InternalServerError([ErrorMessage, .. ex.ExtractMessages()]); } } -} \ No newline at end of file +} diff --git a/src/MaksIT.Dapr/Services/DaprWorkLeaseStore.cs b/src/MaksIT.Dapr/Services/DaprWorkLeaseStore.cs new file mode 100644 index 0000000..0b9087d --- /dev/null +++ b/src/MaksIT.Dapr/Services/DaprWorkLeaseStore.cs @@ -0,0 +1,126 @@ +using MaksIT.Results; + + +namespace MaksIT.Dapr.Services; + +public sealed record DaprWorkLease( + string HolderId, + DateTimeOffset AcquiredAtUtc, + DateTimeOffset ExpiresAtUtc +); + +/// +/// HA work coordination via Dapr state store (broker-agnostic). +/// Keys are product-defined; store name comes from the Dapr Component. +/// +public interface IDaprWorkLeaseStore { + Task> TryAcquireAsync(string storeName, string workKey, string holderId, TimeSpan ttl, CancellationToken cancellationToken = default); + Task> TryRenewAsync(string storeName, string workKey, string holderId, TimeSpan ttl, CancellationToken cancellationToken = default); + Task ReleaseAsync(string storeName, string workKey, string holderId, CancellationToken cancellationToken = default); + Task> GetAsync(string storeName, string workKey, CancellationToken cancellationToken = default); +} + +public sealed class DaprWorkLeaseStore( + IDaprStateStoreService stateStore +) : IDaprWorkLeaseStore { + + public async Task> TryAcquireAsync( + string storeName, + string workKey, + string holderId, + TimeSpan ttl, + CancellationToken cancellationToken = default) { + var validation = Validate(storeName, workKey, holderId, ttl); + if (!validation.IsSuccess) + return validation.ToResultOfType(false); + + var existing = await stateStore.GetStateAndETagAsync(storeName, workKey, cancellationToken).ConfigureAwait(false); + if (!existing.IsSuccess) + return existing.ToResult().ToResultOfType(false); + + var (lease, etag) = existing.Value; + var now = DateTimeOffset.UtcNow; + + if (lease is not null && lease.ExpiresAtUtc > now && !string.Equals(lease.HolderId, holderId, StringComparison.Ordinal)) + return Result.Ok(false); + + var next = new DaprWorkLease(holderId, now, now.Add(ttl)); + // First write: etag may be null/empty when key missing. + var saved = await stateStore.TrySaveStateAsync(storeName, workKey, next, etag, cancellationToken).ConfigureAwait(false); + if (!saved.IsSuccess) + return saved; + + return Result.Ok(saved.Value); + } + + public async Task> TryRenewAsync( + string storeName, + string workKey, + string holderId, + TimeSpan ttl, + CancellationToken cancellationToken = default) { + var validation = Validate(storeName, workKey, holderId, ttl); + if (!validation.IsSuccess) + return validation.ToResultOfType(false); + + var existing = await stateStore.GetStateAndETagAsync(storeName, workKey, cancellationToken).ConfigureAwait(false); + if (!existing.IsSuccess) + return existing.ToResult().ToResultOfType(false); + + var (lease, etag) = existing.Value; + if (lease is null || !string.Equals(lease.HolderId, holderId, StringComparison.Ordinal)) + return Result.Ok(false); + + var now = DateTimeOffset.UtcNow; + var next = lease with { ExpiresAtUtc = now.Add(ttl) }; + var saved = await stateStore.TrySaveStateAsync(storeName, workKey, next, etag, cancellationToken).ConfigureAwait(false); + if (!saved.IsSuccess) + return saved; + + return Result.Ok(saved.Value); + } + + public async Task ReleaseAsync( + string storeName, + string workKey, + string holderId, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || string.IsNullOrWhiteSpace(workKey) || string.IsNullOrWhiteSpace(holderId)) + return Result.BadRequest("storeName, workKey, and holderId are required."); + + var existing = await stateStore.GetStateAndETagAsync(storeName, workKey, cancellationToken).ConfigureAwait(false); + if (!existing.IsSuccess) + return existing.ToResult(); + + var (lease, _) = existing.Value; + if (lease is null) + return Result.Ok(); + + if (!string.Equals(lease.HolderId, holderId, StringComparison.Ordinal)) + return Result.Conflict("Lease is held by another instance."); + + return await stateStore.DeleteStateAsync(storeName, workKey).ConfigureAwait(false); + } + + public async Task> GetAsync( + string storeName, + string workKey, + CancellationToken cancellationToken = default) { + if (string.IsNullOrWhiteSpace(storeName) || string.IsNullOrWhiteSpace(workKey)) + return Result.BadRequest(null, "storeName and workKey are required."); + + var existing = await stateStore.GetStateAndETagAsync(storeName, workKey, cancellationToken).ConfigureAwait(false); + if (!existing.IsSuccess) + return existing.ToResult().ToResultOfType(null); + + return Result.Ok(existing.Value.Value); + } + + private static Result Validate(string storeName, string workKey, string holderId, TimeSpan ttl) { + if (string.IsNullOrWhiteSpace(storeName) || string.IsNullOrWhiteSpace(workKey) || string.IsNullOrWhiteSpace(holderId)) + return Result.BadRequest("storeName, workKey, and holderId are required."); + if (ttl <= TimeSpan.Zero) + return Result.BadRequest("ttl must be positive."); + return Result.Ok(); + } +}