< Summary - AsyncResponse (Release / net8.0+net10.0 / unit+integration)

Information
Class: AsyncResponse.DurableFlows.DynamoDB.DynamoDbFlowStateStore
Assembly: AsyncResponse.DurableFlows.DynamoDB
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.DynamoDB/DynamoDbDurableFlows.cs
Line coverage
99%
Covered lines: 265
Uncovered lines: 2
Coverable lines: 267
Total lines: 505
Line coverage: 99.2%
Branch coverage
87%
Covered branches: 82
Total branches: 94
Branch coverage: 87.2%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
LoadAsync()100%1818100%
TryCreateAsync()100%11100%
TryUpdateAsync()100%22100%
TryAcquireLeaseAsync(...)100%11100%
TryRenewLeaseAsync(...)100%11100%
ReleaseLeaseAsync()100%11100%
TryDeleteAsync()100%22100%
EnsureCreatedAsync()85.71%424298.68%
WaitForTableActiveAsync()75%4488.89%
ValidateTableSchema(...)75%2020100%
UpdateLeaseAsync()100%44100%
CreateItem(...)100%11100%
Key(...)100%11100%
UnixSeconds(...)100%11100%
UnixSecondsCeiling(...)100%11100%
UnixMilliseconds(...)100%11100%
Dispose()100%22100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.DynamoDB/DynamoDbDurableFlows.cs

#LineLine coverage
 1using Amazon.DynamoDBv2;
 2using Amazon.DynamoDBv2.Model;
 3using AsyncResponse;
 4using AsyncResponse.DurableFlows.DynamoDB;
 5using AsyncResponse.DurableFlows.Internal;
 6using Microsoft.Extensions.DependencyInjection.Extensions;
 7using Microsoft.Extensions.Options;
 8using System.Globalization;
 9
 10namespace Microsoft.Extensions.DependencyInjection
 11{
 12    /// <summary>DI registration for the DynamoDB durable-flow state store.</summary>
 13    public static class DynamoDbDurableFlowServiceCollectionExtensions
 14    {
 15        /// <summary>
 16        /// Stores durable-flow state in DynamoDB. Hosts may register an <see cref="IAmazonDynamoDB"/>
 17        /// client; otherwise the default AWS credential/region chain is used.
 18        /// </summary>
 19        public static AsyncResponseRegistrationBuilder WithDynamoDbDurableFlows(
 20            this AsyncResponseRegistrationBuilder builder,
 21            Action<DynamoDbDurableFlowOptions>? configure = null)
 22        {
 23            // Singleton on purpose: table/TTL provisioning is cached per store instance and DynamoDB
 24            // control-plane calls are throttled account-wide — a scoped store would re-issue them on
 25            // every flow execution. A host-registered IAmazonDynamoDB is reused when present;
 26            // otherwise the store creates and owns a client from the default AWS credential/region
 27            // chain. Nothing is registered as a bare IAmazonDynamoDB service, so unrelated
 28            // resolutions of that type are never answered — or broken — by this package.
 29            builder.Services.TryAddSingleton(provider =>
 30            {
 31                var options = provider.GetRequiredService<IOptions<DynamoDbDurableFlowOptions>>();
 32                var shared = provider.GetService<IAmazonDynamoDB>();
 33                return shared is not null
 34                    ? new DynamoDbFlowStateStore(shared, options)
 35                    : new DynamoDbFlowStateStore(new AmazonDynamoDBClient(), options, ownsClient: true);
 36            });
 37            return builder.WithDurableFlows<DynamoDbFlowStateStore, DynamoDbDurableFlowOptions>(configure);
 38        }
 39    }
 40}
 41
 42namespace AsyncResponse.DurableFlows.DynamoDB
 43{
 44/// <summary>Options for the DynamoDB durable-flow state store.</summary>
 45public sealed class DynamoDbDurableFlowOptions : DurableFlowOptions
 46{
 47    /// <summary>Table storing one durable-flow ledger item per flow id.</summary>
 48    public string TableName { get; set; } = "AsyncResponseFlowState";
 49
 50    /// <summary>Creates the table on first use when it does not exist.</summary>
 51    public bool AutoCreateTable { get; set; } = true;
 52
 53    /// <summary>Enables DynamoDB TTL on the expiry attribute when auto-creating the table.</summary>
 54    public bool EnableTimeToLive { get; set; } = true;
 55
 56    /// <summary>Attribute used for DynamoDB TTL. Default: <c>expires_at</c>.</summary>
 57    public string TimeToLiveAttributeName { get; set; } = "expires_at";
 58
 59    /// <summary>
 60    /// Maximum serialized flow-state size in bytes accepted by writes; oversized ledgers fail fast
 61    /// with an actionable error instead of the raw 400 KB item-cap ValidationException the
 62    /// executor would retry into the dead-letter queue. Default: 350 KB (headroom under DynamoDB's
 63    /// 400 KB item cap for the sibling attributes); <c>null</c> disables the guard.
 64    /// </summary>
 65    public long? MaxStateBytes { get; set; } = 350_000;
 66
 67    /// <summary>Validates option values and throws on misconfiguration.</summary>
 68    public void Validate()
 69    {
 70        if (string.IsNullOrWhiteSpace(TableName))
 71            throw new InvalidOperationException($"{nameof(DynamoDbDurableFlowOptions)}.{nameof(TableName)} must be confi
 72        if (string.IsNullOrWhiteSpace(TimeToLiveAttributeName))
 73            throw new InvalidOperationException($"{nameof(DynamoDbDurableFlowOptions)}.{nameof(TimeToLiveAttributeName)}
 74        if (MaxStateBytes is <= 0)
 75            throw new InvalidOperationException($"{nameof(DynamoDbDurableFlowOptions)}.{nameof(MaxStateBytes)} must be p
 76    }
 77}
 78
 79/// <summary>DynamoDB implementation of <see cref="IFlowStateStore"/>.</summary>
 80public sealed class DynamoDbFlowStateStore : IFlowStateStore, IDisposable
 81{
 82    private const string FlowIdAttribute = "flow_id";
 83    private const string StateJsonAttribute = "state_json";
 84    private const string UpdatedAtAttribute = "updated_at";
 85    private const string RevisionAttribute = "revision";
 86    private const string LeaseIdAttribute = "lease_id";
 87    private const string LeaseExpiresAtAttribute = "lease_expires_at_ms";
 88
 89    // Time authority: this store keeps the app clock (DateTimeOffset.UtcNow) for expiry and lease
 90    // comparisons. DynamoDB condition expressions evaluate client-supplied values only — there is
 91    // no server-clock function available in a conditional write — so multi-node deployments
 92    // should keep worker clocks synchronized well inside the lease window. (DynamoDB's own TTL
 93    // reaper, by contrast, runs on the service clock against the epoch-seconds expiry attribute.)
 94
 95    private readonly IAmazonDynamoDB _client;
 96    private readonly DynamoDbDurableFlowOptions _options;
 397    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 98    private readonly bool _ownsClient;
 99    private bool _created;
 100
 3101    public DynamoDbFlowStateStore(IAmazonDynamoDB client, IOptions<DynamoDbDurableFlowOptions> options, bool ownsClient 
 102    {
 3103        _client = client;
 3104        _options = options.Value;
 3105        _options.Validate();
 3106        _ownsClient = ownsClient;
 3107    }
 108
 109    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 110    {
 3111        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 3112        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 113
 3114        var response = await _client.GetItemAsync(new GetItemRequest
 3115        {
 3116            TableName = _options.TableName,
 3117            Key = Key(flowId),
 3118            ConsistentRead = true
 3119        }, cancellationToken).ConfigureAwait(false);
 120
 3121        if (response.Item is null || response.Item.Count == 0)
 3122            return null;
 3123        if (!response.Item.TryGetValue(_options.TimeToLiveAttributeName, out var expires) ||
 3124            !long.TryParse(expires.N, NumberStyles.Integer, CultureInfo.InvariantCulture, out var expiresAt) ||
 3125            expiresAt <= DateTimeOffset.UtcNow.ToUnixTimeSeconds())
 3126            return null;
 3127        if (!response.Item.TryGetValue(StateJsonAttribute, out var json) || string.IsNullOrEmpty(json.S))
 2128            return null;
 129
 3130        if (!response.Item.TryGetValue(RevisionAttribute, out var revision)
 3131            || !long.TryParse(revision.N, NumberStyles.Integer, CultureInfo.InvariantCulture, out var value))
 3132            return null;
 133
 3134        return DurableFlowStoreShared.ReadState(flowId, json.S, value);
 3135    }
 136
 137    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 138    {
 3139        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 3140        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "DynamoDB");
 3141        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 142
 3143        var now = DateTimeOffset.UtcNow;
 144        try
 145        {
 3146            await _client.PutItemAsync(new PutItemRequest
 3147            {
 3148                TableName = _options.TableName,
 3149                Item = CreateItem(flowId, stateJson, state.Revision, ttl, now),
 3150                ConditionExpression = "attribute_not_exists(#flow_id) OR #expires <= :now",
 3151                ExpressionAttributeNames = new Dictionary<string, string>
 3152                {
 3153                    ["#flow_id"] = FlowIdAttribute,
 3154                    ["#expires"] = _options.TimeToLiveAttributeName
 3155                },
 3156                ExpressionAttributeValues = new Dictionary<string, AttributeValue>
 3157                {
 3158                    [":now"] = new() { N = UnixSeconds(now) }
 3159                }
 3160            }, cancellationToken).ConfigureAwait(false);
 3161            return true;
 162        }
 3163        catch (ConditionalCheckFailedException)
 164        {
 3165            return false;
 166        }
 3167    }
 168
 169    public async Task<bool> TryUpdateAsync(
 170        string flowId,
 171        FlowState state,
 172        long expectedRevision,
 173        TimeSpan ttl,
 174        string? leaseId = null,
 175        CancellationToken cancellationToken = default)
 176    {
 3177        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 3178        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "DynamoDB");
 3179        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 180
 3181        var now = DateTimeOffset.UtcNow;
 3182        var names = new Dictionary<string, string>
 3183        {
 3184            ["#state"] = StateJsonAttribute,
 3185            ["#expires"] = _options.TimeToLiveAttributeName,
 3186            ["#updated"] = UpdatedAtAttribute,
 3187            ["#revision"] = RevisionAttribute
 3188        };
 3189        var values = new Dictionary<string, AttributeValue>
 3190        {
 3191            [":state"] = new() { S = stateJson },
 3192            [":expires"] = new() { N = UnixSecondsCeiling(DurableFlowStoreShared.AddSaturating(now, ttl)) },
 3193            [":updated"] = new() { N = UnixSeconds(now) },
 3194            [":expected_revision"] = new() { N = expectedRevision.ToString(CultureInfo.InvariantCulture) },
 3195            [":new_revision"] = new() { N = state.Revision.ToString(CultureInfo.InvariantCulture) },
 3196            [":now"] = new() { N = UnixSeconds(now) }
 3197        };
 3198        var condition = "#revision = :expected_revision AND #expires > :now";
 3199        if (leaseId is not null)
 200        {
 3201            condition += " AND #lease_id = :lease_id AND #lease_expires > :now_ms";
 3202            names["#lease_id"] = LeaseIdAttribute;
 3203            names["#lease_expires"] = LeaseExpiresAtAttribute;
 3204            values[":lease_id"] = new AttributeValue { S = leaseId };
 3205            values[":now_ms"] = new AttributeValue { N = UnixMilliseconds(now) };
 206        }
 207
 208        try
 209        {
 3210            await _client.UpdateItemAsync(new UpdateItemRequest
 3211            {
 3212                TableName = _options.TableName,
 3213                Key = Key(flowId),
 3214                UpdateExpression = "SET #state = :state, #expires = :expires, #updated = :updated, #revision = :new_revi
 3215                ConditionExpression = condition,
 3216                ExpressionAttributeNames = names,
 3217                ExpressionAttributeValues = values
 3218            }, cancellationToken).ConfigureAwait(false);
 3219            return true;
 220        }
 3221        catch (ConditionalCheckFailedException)
 222        {
 3223            return false;
 224        }
 3225    }
 226
 227    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 3228        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 229
 230    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 3231        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 232
 233    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 234    {
 3235        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 236        try
 237        {
 3238            await _client.UpdateItemAsync(new UpdateItemRequest
 3239            {
 3240                TableName = _options.TableName,
 3241                Key = Key(flowId),
 3242                UpdateExpression = "REMOVE #lease_id, #lease_expires",
 3243                ConditionExpression = "#lease_id = :lease_id",
 3244                ExpressionAttributeNames = new Dictionary<string, string>
 3245                {
 3246                    ["#lease_id"] = LeaseIdAttribute,
 3247                    ["#lease_expires"] = LeaseExpiresAtAttribute
 3248                },
 3249                ExpressionAttributeValues = new Dictionary<string, AttributeValue>
 3250                {
 3251                    [":lease_id"] = new() { S = leaseId }
 3252                }
 3253            }, cancellationToken).ConfigureAwait(false);
 3254        }
 3255        catch (ConditionalCheckFailedException)
 256        {
 3257        }
 3258    }
 259
 260    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 261    {
 3262        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 3263        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 264
 3265        var response = await _client.DeleteItemAsync(new DeleteItemRequest
 3266        {
 3267            TableName = _options.TableName,
 3268            Key = Key(flowId),
 3269            ReturnValues = ReturnValue.ALL_OLD
 3270        }, cancellationToken).ConfigureAwait(false);
 3271        return response.Attributes is { Count: > 0 };
 3272    }
 273
 274    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 275    {
 3276        if (_created)
 3277            return;
 278
 3279        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 280        try
 281        {
 3282            if (_created)
 0283                return;
 284
 3285            TableDescription? table = null;
 286            try
 287            {
 3288                var described = await _client.DescribeTableAsync(_options.TableName, cancellationToken).ConfigureAwait(f
 2289                table = described.Table;
 2290            }
 3291            catch (ResourceNotFoundException)
 292            {
 3293            }
 294
 3295            if (table is null)
 296            {
 3297                if (!_options.AutoCreateTable)
 2298                    throw new InvalidOperationException(
 2299                        $"DynamoDB table '{_options.TableName}' does not exist and {nameof(DynamoDbDurableFlowOptions.Au
 300
 301                try
 302                {
 3303                    await _client.CreateTableAsync(new CreateTableRequest
 3304                    {
 3305                        TableName = _options.TableName,
 3306                        BillingMode = BillingMode.PAY_PER_REQUEST,
 3307                        AttributeDefinitions =
 3308                        [
 3309                            new AttributeDefinition(FlowIdAttribute, ScalarAttributeType.S)
 3310                        ],
 3311                        KeySchema =
 3312                        [
 3313                            new KeySchemaElement(FlowIdAttribute, KeyType.HASH)
 3314                        ]
 3315                    }, cancellationToken).ConfigureAwait(false);
 3316                }
 2317                catch (ResourceInUseException)
 318                {
 319                    // Another process won the create race; fall through and wait for ACTIVE.
 2320                }
 321
 3322                table = await WaitForTableActiveAsync(cancellationToken).ConfigureAwait(false);
 323            }
 2324            else if (table.TableStatus != TableStatus.ACTIVE)
 325            {
 2326                table = await WaitForTableActiveAsync(cancellationToken).ConfigureAwait(false);
 327            }
 328
 3329            ValidateTableSchema(table);
 330
 3331            if (_options.EnableTimeToLive)
 332            {
 333                // Check the TTL status instead of blind-enabling: UpdateTimeToLive throws when TTL
 334                // is already enabled, and relying on a swallowed exception per provisioning is noise.
 3335                var ttlStatus = await _client.DescribeTimeToLiveAsync(new DescribeTimeToLiveRequest
 3336                {
 3337                    TableName = _options.TableName
 3338                }, cancellationToken).ConfigureAwait(false);
 339
 3340                var description = ttlStatus.TimeToLiveDescription;
 3341                var status = description?.TimeToLiveStatus;
 3342                if ((status == TimeToLiveStatus.ENABLED || status == TimeToLiveStatus.ENABLING)
 3343                    && !string.Equals(description?.AttributeName, _options.TimeToLiveAttributeName, StringComparison.Ord
 344                {
 2345                    throw new InvalidOperationException(
 2346                        $"DynamoDB table '{_options.TableName}' has TTL configured on attribute '{description?.Attribute
 2347                        $"but '{_options.TimeToLiveAttributeName}' is required.");
 348                }
 349
 3350                if (status != TimeToLiveStatus.ENABLED && status != TimeToLiveStatus.ENABLING)
 351                {
 3352                    if (!_options.AutoCreateTable)
 353                    {
 2354                        throw new InvalidOperationException(
 2355                            $"DynamoDB table '{_options.TableName}' does not have TTL enabled on " +
 2356                            $"'{_options.TimeToLiveAttributeName}'. Enable it in infrastructure before using the table."
 357                    }
 358
 359                    try
 360                    {
 3361                        await _client.UpdateTimeToLiveAsync(new UpdateTimeToLiveRequest
 3362                        {
 3363                            TableName = _options.TableName,
 3364                            TimeToLiveSpecification = new TimeToLiveSpecification
 3365                            {
 3366                                AttributeName = _options.TimeToLiveAttributeName,
 3367                                Enabled = true
 3368                            }
 3369                        }, cancellationToken).ConfigureAwait(false);
 3370                    }
 2371                    catch (AmazonDynamoDBException ex) when (string.Equals(ex.ErrorCode, "ValidationException", StringCo
 372                    {
 373                        // Accept only the one safe race: another process enabled the expected TTL
 374                        // attribute between our describe and update. Do not swallow unrelated
 375                        // validation failures.
 2376                        var raced = await _client.DescribeTimeToLiveAsync(new DescribeTimeToLiveRequest
 2377                        {
 2378                            TableName = _options.TableName
 2379                        }, cancellationToken).ConfigureAwait(false);
 2380                        var racedDescription = raced.TimeToLiveDescription;
 2381                        if ((racedDescription?.TimeToLiveStatus != TimeToLiveStatus.ENABLED
 2382                                && racedDescription?.TimeToLiveStatus != TimeToLiveStatus.ENABLING)
 2383                            || !string.Equals(racedDescription.AttributeName, _options.TimeToLiveAttributeName, StringCo
 384                        {
 2385                            throw;
 386                        }
 387                    }
 388                }
 389            }
 390
 3391            _created = true;
 3392        }
 393        finally
 394        {
 3395            _ensureGate.Release();
 396        }
 3397    }
 398
 399    private async Task<TableDescription> WaitForTableActiveAsync(CancellationToken cancellationToken)
 400    {
 3401        var deadline = DateTime.UtcNow.AddSeconds(30);
 2402        while (true)
 403        {
 3404            var response = await _client.DescribeTableAsync(_options.TableName, cancellationToken).ConfigureAwait(false)
 3405            if (response.Table.TableStatus == TableStatus.ACTIVE)
 3406                return response.Table;
 2407            if (DateTime.UtcNow >= deadline)
 0408                throw new TimeoutException($"DynamoDB table '{_options.TableName}' did not become ACTIVE within 30 secon
 409
 2410            await Task.Delay(TimeSpan.FromMilliseconds(250), cancellationToken).ConfigureAwait(false);
 411        }
 3412    }
 413
 414    private void ValidateTableSchema(TableDescription table)
 415    {
 3416        var hashKey = table.KeySchema?.SingleOrDefault(key => key.KeyType == KeyType.HASH);
 3417        var keyDefinition = table.AttributeDefinitions?
 3418            .SingleOrDefault(attribute => string.Equals(attribute.AttributeName, FlowIdAttribute, StringComparison.Ordin
 3419        if (table.KeySchema?.Count != 1
 3420            || !string.Equals(hashKey?.AttributeName, FlowIdAttribute, StringComparison.Ordinal)
 3421            || keyDefinition?.AttributeType != ScalarAttributeType.S)
 422        {
 2423            throw new InvalidOperationException(
 2424                $"DynamoDB table '{_options.TableName}' must use one string partition key named '{FlowIdAttribute}' and 
 425        }
 3426    }
 427
 428    private async Task<bool> UpdateLeaseAsync(
 429        string flowId,
 430        string leaseId,
 431        TimeSpan leaseDuration,
 432        bool acquire,
 433        CancellationToken cancellationToken)
 434    {
 3435        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 3436        ArgumentException.ThrowIfNullOrWhiteSpace(leaseId);
 3437        if (leaseDuration <= TimeSpan.Zero)
 2438            throw new ArgumentOutOfRangeException(nameof(leaseDuration));
 439
 3440        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3441        var now = DateTimeOffset.UtcNow;
 442        try
 443        {
 3444            await _client.UpdateItemAsync(new UpdateItemRequest
 3445            {
 3446                TableName = _options.TableName,
 3447                Key = Key(flowId),
 3448                UpdateExpression = "SET #lease_id = :lease_id, #lease_expires = :lease_expires",
 3449                ConditionExpression = acquire
 3450                    ? "#expires > :now AND attribute_exists(#revision) AND (attribute_not_exists(#lease_id) OR #lease_ex
 3451                    : "#expires > :now AND attribute_exists(#revision) AND #lease_id = :lease_id AND #lease_expires > :n
 3452                ExpressionAttributeNames = new Dictionary<string, string>
 3453                {
 3454                    ["#expires"] = _options.TimeToLiveAttributeName,
 3455                    ["#revision"] = RevisionAttribute,
 3456                    ["#lease_id"] = LeaseIdAttribute,
 3457                    ["#lease_expires"] = LeaseExpiresAtAttribute
 3458                },
 3459                ExpressionAttributeValues = new Dictionary<string, AttributeValue>
 3460                {
 3461                    [":now"] = new() { N = UnixSeconds(now) },
 3462                    [":now_ms"] = new() { N = UnixMilliseconds(now) },
 3463                    [":lease_id"] = new() { S = leaseId },
 3464                    [":lease_expires"] = new() { N = UnixMilliseconds(DurableFlowStoreShared.AddSaturating(now, leaseDur
 3465                }
 3466            }, cancellationToken).ConfigureAwait(false);
 3467            return true;
 468        }
 3469        catch (ConditionalCheckFailedException)
 470        {
 3471            return false;
 472        }
 3473    }
 474
 475    private Dictionary<string, AttributeValue> CreateItem(string flowId, string stateJson, long revision, TimeSpan ttl, 
 3476        => new(StringComparer.Ordinal)
 3477        {
 3478            [FlowIdAttribute] = new() { S = flowId },
 3479            [StateJsonAttribute] = new() { S = stateJson },
 3480            [_options.TimeToLiveAttributeName] = new() { N = UnixSecondsCeiling(DurableFlowStoreShared.AddSaturating(now
 3481            [UpdatedAtAttribute] = new() { N = UnixSeconds(now) },
 3482            [RevisionAttribute] = new() { N = revision.ToString(CultureInfo.InvariantCulture) }
 3483        };
 484
 485    private static Dictionary<string, AttributeValue> Key(string flowId)
 3486        => new(StringComparer.Ordinal) { [FlowIdAttribute] = new AttributeValue { S = flowId } };
 487
 488    private static string UnixSeconds(DateTimeOffset value)
 3489        => value.ToUnixTimeSeconds().ToString(CultureInfo.InvariantCulture);
 490
 491    private static string UnixSecondsCeiling(DateTimeOffset value)
 3492        => ((long)Math.Ceiling(value.ToUnixTimeMilliseconds() / 1000.0)).ToString(CultureInfo.InvariantCulture);
 493
 494    private static string UnixMilliseconds(DateTimeOffset value)
 3495        => value.ToUnixTimeMilliseconds().ToString(CultureInfo.InvariantCulture);
 496
 497    /// <summary>Disposes the DynamoDB client when the store created (and therefore owns) it.</summary>
 498    public void Dispose()
 499    {
 2500        _ensureGate.Dispose();
 2501        if (_ownsClient)
 2502            _client.Dispose();
 2503    }
 504}
 505}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(Amazon.DynamoDBv2.IAmazonDynamoDB, Microsoft.Extensions.Options.IOptions<AsyncResponse.DurableFlows.DynamoDB.DynamoDbDurableFlowOptions>, bool)
LoadAsync()
TryCreateAsync()
TryUpdateAsync()
TryAcquireLeaseAsync(string, string, System.TimeSpan, System.Threading.CancellationToken)
TryRenewLeaseAsync(string, string, System.TimeSpan, System.Threading.CancellationToken)
ReleaseLeaseAsync()
TryDeleteAsync()
EnsureCreatedAsync()
WaitForTableActiveAsync()
ValidateTableSchema(Amazon.DynamoDBv2.Model.TableDescription)
UpdateLeaseAsync()
CreateItem(string, string, long, System.TimeSpan, System.DateTimeOffset)
Key(string)
UnixSeconds(System.DateTimeOffset)
UnixSecondsCeiling(System.DateTimeOffset)
UnixMilliseconds(System.DateTimeOffset)
Dispose()