(feature): add HA work leases and pub/sub helpers for v2.1.0
This commit is contained in:
parent
d55e3af7d2
commit
89183c67b9
36
CHANGELOG.md
36
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<T>`, `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.
|
||||
|
||||
<!--
|
||||
Template for new releases:
|
||||
|
||||
## v1.x.x
|
||||
|
||||
### Added
|
||||
- New features
|
||||
|
||||
### Changed
|
||||
- Changes in existing functionality
|
||||
|
||||
### Deprecated
|
||||
- Soon-to-be removed features
|
||||
|
||||
### Removed
|
||||
- Removed features
|
||||
|
||||
### Fixed
|
||||
- Bug fixes
|
||||
|
||||
### Security
|
||||
- Security improvements
|
||||
-->
|
||||
|
||||
30
README.md
30
README.md
@ -1,8 +1,8 @@
|
||||
# MaksIT.Dapr
|
||||
|
||||

|
||||

|
||||

|
||||

|
||||

|
||||

|
||||
|
||||
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<T>` and map with `DaprPubSubAck.ToActionResult` (2xx ACK, 503 Busy → redelivery).
|
||||
|
||||
## Contributing
|
||||
|
||||
|
||||
82
src/MaksIT.Dapr.Tests/DaprWorkLeaseStoreTests.cs
Normal file
82
src/MaksIT.Dapr.Tests/DaprWorkLeaseStoreTests.cs
Normal file
@ -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<IDaprStateStoreService>();
|
||||
state
|
||||
.Setup(s => s.GetStateAndETagAsync<DaprWorkLease>("store", "work", It.IsAny<CancellationToken>()))
|
||||
.ReturnsAsync(MaksIT.Results.Result<(DaprWorkLease? Value, string? ETag)>.Ok((null, null)));
|
||||
state
|
||||
.Setup(s => s.TrySaveStateAsync("store", "work", It.IsAny<DaprWorkLease>(), null, It.IsAny<CancellationToken>()))
|
||||
.ReturnsAsync(MaksIT.Results.Result<bool>.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<IDaprStateStoreService>();
|
||||
state
|
||||
.Setup(s => s.GetStateAndETagAsync<DaprWorkLease>("store", "work", It.IsAny<CancellationToken>()))
|
||||
.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<ObjectResult>(result);
|
||||
Assert.Equal(StatusCodes.Status503ServiceUnavailable, objectResult.StatusCode);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ToActionResult_Accepted_Returns200() {
|
||||
var result = DaprPubSubAck.ToActionResult(DaprPubSubAcceptResult.Accepted());
|
||||
Assert.IsType<OkObjectResult>(result);
|
||||
}
|
||||
}
|
||||
|
||||
public class DaprStateStoreETagTests {
|
||||
[Fact]
|
||||
public async Task TrySaveStateAsync_ReturnsOkFalse_WhenClientReturnsFalse() {
|
||||
var clientMock = new Mock<DaprClient>();
|
||||
clientMock
|
||||
.Setup(x => x.TrySaveStateAsync(
|
||||
It.IsAny<string>(),
|
||||
It.IsAny<string>(),
|
||||
It.IsAny<string>(),
|
||||
It.IsAny<string>(),
|
||||
It.IsAny<StateOptions>(),
|
||||
It.IsAny<IReadOnlyDictionary<string, string>>(),
|
||||
It.IsAny<CancellationToken>()))
|
||||
.ReturnsAsync(false);
|
||||
|
||||
var service = new DaprStateStoreService(Mock.Of<ILogger<DaprStateStoreService>>(), clientMock.Object);
|
||||
var result = await service.TrySaveStateAsync("store", "key", "value", "etag");
|
||||
|
||||
Assert.True(result.IsSuccess);
|
||||
Assert.False(result.Value);
|
||||
}
|
||||
}
|
||||
@ -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<IDaprStateStoreService, DaprStateStoreService>();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Registers Dapr state store plus HA work-lease coordination and runtime instance id.
|
||||
/// </summary>
|
||||
public static void RegisterWorkLeases(this IServiceCollection services) {
|
||||
services.RegisterStateStore();
|
||||
services.AddSingleton<IDaprRuntimeInstanceId, DaprRuntimeInstanceIdProvider>();
|
||||
services.AddSingleton<IDaprWorkLeaseStore, DaprWorkLeaseStore>();
|
||||
}
|
||||
}
|
||||
|
||||
@ -7,7 +7,7 @@
|
||||
|
||||
<!-- NuGet package metadata -->
|
||||
<PackageId>MaksIT.Dapr</PackageId>
|
||||
<Version>2.0.1</Version>
|
||||
<Version>2.1.0</Version>
|
||||
<Authors>Maksym Sadovnychyy</Authors>
|
||||
<Company>MAKS-IT</Company>
|
||||
<Product>MaksIT.Dapr</Product>
|
||||
|
||||
45
src/MaksIT.Dapr/PubSub/DaprPubSubWork.cs
Normal file
45
src/MaksIT.Dapr/PubSub/DaprPubSubWork.cs
Normal file
@ -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);
|
||||
}
|
||||
|
||||
/// <summary>Product implements accept/claim; HTTP layer maps to Dapr ACK/NAK via <see cref="DaprPubSubAck"/>.</summary>
|
||||
public interface IDaprPubSubWorkHandler<TMessage> {
|
||||
Task<DaprPubSubAcceptResult> TryAcceptAsync(TMessage message, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
/// <summary>Maps accept outcomes to HTTP status codes for Dapr pub/sub delivery.</summary>
|
||||
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),
|
||||
};
|
||||
}
|
||||
26
src/MaksIT.Dapr/Services/DaprRuntimeInstanceIdProvider.cs
Normal file
26
src/MaksIT.Dapr/Services/DaprRuntimeInstanceIdProvider.cs
Normal file
@ -0,0 +1,26 @@
|
||||
namespace MaksIT.Dapr.Services;
|
||||
|
||||
/// <summary>Stable id for this process/pod (lease holder).</summary>
|
||||
public interface IDaprRuntimeInstanceId {
|
||||
string InstanceId { get; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Prefers <c>POD_NAME</c> in Kubernetes; otherwise host name + process id.
|
||||
/// </summary>
|
||||
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}";
|
||||
}
|
||||
}
|
||||
@ -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<Result> SetStateAsync<T>(string storeName, string key, T value);
|
||||
Task<Result<T?>> GetStateAsync<T>(string storeName, string key);
|
||||
Task<Result<(T? Value, string? ETag)>> GetStateAndETagAsync<T>(string storeName, string key, CancellationToken cancellationToken = default);
|
||||
Task<Result<bool>> TrySaveStateAsync<T>(string storeName, string key, T value, string? etag, CancellationToken cancellationToken = default);
|
||||
Task<Result> 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<DaprStateStoreService> _logger;
|
||||
|
||||
public DaprStateStoreService(
|
||||
ILogger<DaprStateStoreService> logger,
|
||||
DaprClient client
|
||||
) {
|
||||
public DaprStateStoreService(ILogger<DaprStateStoreService> logger, DaprClient client) {
|
||||
_logger = logger;
|
||||
_client = client;
|
||||
}
|
||||
/// <summary>
|
||||
/// Saves a state to a Dapr state store
|
||||
/// </summary>
|
||||
/// <typeparam name="T"></typeparam>
|
||||
/// <param name="storeName"></param>
|
||||
/// <param name="key"></param>
|
||||
/// <param name="value"></param>
|
||||
/// <returns></returns>
|
||||
|
||||
public async Task<Result> SetStateAsync<T>(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()]);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets a state from a Dapr state store
|
||||
/// </summary>
|
||||
/// <typeparam name="T"></typeparam>
|
||||
/// <param name="storeName"></param>
|
||||
/// <param name="key"></param>
|
||||
/// <returns></returns>
|
||||
public async Task<Result<T?>> GetStateAsync<T>(string storeName, string key) {
|
||||
try {
|
||||
var state = await _client.GetStateAsync<T?>(storeName, key);
|
||||
if (state == null)
|
||||
if (state is null)
|
||||
return Result<T?>.NotFound(default, $"State from the store {storeName} with the {key} not found.");
|
||||
|
||||
return Result<T?>.Ok(state);
|
||||
}
|
||||
catch (Exception ex) {
|
||||
_logger.LogError(ex, _errorMessage);
|
||||
return Result<T?>.InternalServerError(default, new[] {_errorMessage}.Concat(ex.ExtractMessages()).ToArray());
|
||||
_logger.LogError(ex, ErrorMessage);
|
||||
return Result<T?>.InternalServerError(default, [ErrorMessage, .. ex.ExtractMessages()]);
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<Result<(T? Value, string? ETag)>> GetStateAndETagAsync<T>(
|
||||
string storeName,
|
||||
string key,
|
||||
CancellationToken cancellationToken = default) {
|
||||
try {
|
||||
var (value, etag) = await _client.GetStateAndETagAsync<T?>(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<Result<bool>> TrySaveStateAsync<T>(
|
||||
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<bool>.Ok(saved);
|
||||
}
|
||||
catch (Exception ex) {
|
||||
_logger.LogError(ex, ErrorMessage);
|
||||
return Result<bool>.InternalServerError(false, [ErrorMessage, .. ex.ExtractMessages()]);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Deletes a state from a Dapr state store
|
||||
/// </summary>
|
||||
/// <param name="storeName"></param>
|
||||
/// <param name="key"></param>
|
||||
/// <returns></returns>
|
||||
public async Task<Result> 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()]);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
126
src/MaksIT.Dapr/Services/DaprWorkLeaseStore.cs
Normal file
126
src/MaksIT.Dapr/Services/DaprWorkLeaseStore.cs
Normal file
@ -0,0 +1,126 @@
|
||||
using MaksIT.Results;
|
||||
|
||||
|
||||
namespace MaksIT.Dapr.Services;
|
||||
|
||||
public sealed record DaprWorkLease(
|
||||
string HolderId,
|
||||
DateTimeOffset AcquiredAtUtc,
|
||||
DateTimeOffset ExpiresAtUtc
|
||||
);
|
||||
|
||||
/// <summary>
|
||||
/// HA work coordination via Dapr state store (broker-agnostic).
|
||||
/// Keys are product-defined; store name comes from the Dapr Component.
|
||||
/// </summary>
|
||||
public interface IDaprWorkLeaseStore {
|
||||
Task<Result<bool>> TryAcquireAsync(string storeName, string workKey, string holderId, TimeSpan ttl, CancellationToken cancellationToken = default);
|
||||
Task<Result<bool>> TryRenewAsync(string storeName, string workKey, string holderId, TimeSpan ttl, CancellationToken cancellationToken = default);
|
||||
Task<Result> ReleaseAsync(string storeName, string workKey, string holderId, CancellationToken cancellationToken = default);
|
||||
Task<Result<DaprWorkLease?>> GetAsync(string storeName, string workKey, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
public sealed class DaprWorkLeaseStore(
|
||||
IDaprStateStoreService stateStore
|
||||
) : IDaprWorkLeaseStore {
|
||||
|
||||
public async Task<Result<bool>> 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<bool>(false);
|
||||
|
||||
var existing = await stateStore.GetStateAndETagAsync<DaprWorkLease>(storeName, workKey, cancellationToken).ConfigureAwait(false);
|
||||
if (!existing.IsSuccess)
|
||||
return existing.ToResult().ToResultOfType<bool>(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<bool>.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<bool>.Ok(saved.Value);
|
||||
}
|
||||
|
||||
public async Task<Result<bool>> 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<bool>(false);
|
||||
|
||||
var existing = await stateStore.GetStateAndETagAsync<DaprWorkLease>(storeName, workKey, cancellationToken).ConfigureAwait(false);
|
||||
if (!existing.IsSuccess)
|
||||
return existing.ToResult().ToResultOfType<bool>(false);
|
||||
|
||||
var (lease, etag) = existing.Value;
|
||||
if (lease is null || !string.Equals(lease.HolderId, holderId, StringComparison.Ordinal))
|
||||
return Result<bool>.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<bool>.Ok(saved.Value);
|
||||
}
|
||||
|
||||
public async Task<Result> 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<DaprWorkLease>(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<Result<DaprWorkLease?>> GetAsync(
|
||||
string storeName,
|
||||
string workKey,
|
||||
CancellationToken cancellationToken = default) {
|
||||
if (string.IsNullOrWhiteSpace(storeName) || string.IsNullOrWhiteSpace(workKey))
|
||||
return Result<DaprWorkLease?>.BadRequest(null, "storeName and workKey are required.");
|
||||
|
||||
var existing = await stateStore.GetStateAndETagAsync<DaprWorkLease>(storeName, workKey, cancellationToken).ConfigureAwait(false);
|
||||
if (!existing.IsSuccess)
|
||||
return existing.ToResult().ToResultOfType<DaprWorkLease?>(null);
|
||||
|
||||
return Result<DaprWorkLease?>.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();
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user