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

Information
Class: AsyncResponse.DurableFlows.DynamoDB.DynamoDbFlowStateStore
Assembly: AsyncResponse.DurableFlows.DynamoDB
File(s): /_/src/DurableFlows/AsyncResponse.DurableFlows.DynamoDB/DynamoDbDurableFlows.cs
Line coverage
99%
Covered lines: 297
Uncovered lines: 1
Coverable lines: 298
Total lines: 597
Line coverage: 99.6%
Branch coverage
89%
Covered branches: 95
Total branches: 106
Branch coverage: 89.6%
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%
ValidateCreate(...)50%22100%
TryCreateAsync()100%11100%
TryUpdateAsync()100%22100%
TryAcquireLeaseAsync(...)100%11100%
TryRenewLeaseAsync(...)100%11100%
ReleaseLeaseAsync()100%11100%
ObserveLeaseAsync()100%1414100%
.cctor()100%11100%
TryDeleteAsync()100%22100%
EnsureCreatedAsync()88.63%444498.68%
WaitForTableActiveAsync()75%4488.88%
ValidateTableSchema(...)68.75%1616100%
UpdateLeaseAsync()100%22100%
CreateItem(...)100%11100%
Key(...)100%11100%
UnixSeconds(...)100%11100%
UnixSecondsCeiling(...)100%11100%
UnixMilliseconds(...)100%11100%
Dispose()100%22100%

File(s)

/_/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>
 54    /// Enables DynamoDB TTL on the expiry attribute when auto-creating the table. With
 55    /// <see cref="AutoCreateTable"/> off, the store still verifies at startup that the
 56    /// operator-provisioned table has TTL enabled on the expiry attribute regardless of this
 57    /// flag — TTL is the store's only cleanup mechanism.
 58    /// </summary>
 59    public bool EnableTimeToLive { get; set; } = true;
 60
 61    /// <summary>Attribute used for DynamoDB TTL. Default: <c>expires_at</c>.</summary>
 62    public string TimeToLiveAttributeName { get; set; } = "expires_at";
 63
 64    /// <summary>
 65    /// Maximum serialized flow-state size in bytes accepted by writes; oversized ledgers fail fast
 66    /// with an actionable error instead of the raw 400 KB item-cap ValidationException the
 67    /// executor would retry into the dead-letter queue. Default: 350 KB (headroom under DynamoDB's
 68    /// 400 KB item cap for the sibling attributes); <c>null</c> disables the guard.
 69    /// </summary>
 70    public long? MaxStateBytes { get; set; } = 350_000;
 71
 72    /// <summary>Validates option values and throws on misconfiguration.</summary>
 73    public void Validate()
 74    {
 75        if (string.IsNullOrWhiteSpace(TableName))
 76            throw new InvalidOperationException($"{nameof(DynamoDbDurableFlowOptions)}.{nameof(TableName)} must be confi
 77        if (string.IsNullOrWhiteSpace(TimeToLiveAttributeName))
 78            throw new InvalidOperationException($"{nameof(DynamoDbDurableFlowOptions)}.{nameof(TimeToLiveAttributeName)}
 79        // A TTL attribute colliding with one of the store's own attributes would let the later
 80        // item-initializer assignment silently overwrite the TTL value (e.g. with the revision
 81        // number), making every new flow read back as already-expired with no error anywhere.
 82        if (TimeToLiveAttributeName is DynamoDbFlowStateStore.FlowIdAttribute
 83            or DynamoDbFlowStateStore.StateJsonAttribute
 84            or DynamoDbFlowStateStore.UpdatedAtAttribute
 85            or DynamoDbFlowStateStore.RevisionAttribute
 86            or DynamoDbFlowStateStore.LeaseIdAttribute
 87            or DynamoDbFlowStateStore.LeaseExpiresAtAttribute)
 88        {
 89            throw new InvalidOperationException(
 90                $"{nameof(DynamoDbDurableFlowOptions)}.{nameof(TimeToLiveAttributeName)} must not collide with one of th
 91                $"store's own attributes ('{DynamoDbFlowStateStore.FlowIdAttribute}', '{DynamoDbFlowStateStore.StateJson
 92                $"'{DynamoDbFlowStateStore.UpdatedAtAttribute}', '{DynamoDbFlowStateStore.RevisionAttribute}', " +
 93                $"'{DynamoDbFlowStateStore.LeaseIdAttribute}', '{DynamoDbFlowStateStore.LeaseExpiresAtAttribute}').");
 94        }
 95        DurableFlowStoreShared.ValidateMaxStateBytes(MaxStateBytes, nameof(DynamoDbDurableFlowOptions));
 96    }
 97}
 98
 99/// <summary>DynamoDB implementation of <see cref="IFlowStateStore"/>.</summary>
 100public sealed class DynamoDbFlowStateStore : IFlowStateStore, IDisposable
 101{
 102    internal const string FlowIdAttribute = "flow_id";
 103    internal const string StateJsonAttribute = "state_json";
 104    internal const string UpdatedAtAttribute = "updated_at";
 105    internal const string RevisionAttribute = "revision";
 106    internal const string LeaseIdAttribute = "lease_id";
 107    internal const string LeaseExpiresAtAttribute = "lease_expires_at_ms";
 108
 109    // Time authority: this store keeps the app clock (DateTimeOffset.UtcNow) for expiry and lease
 110    // comparisons. DynamoDB condition expressions evaluate client-supplied values only — there is
 111    // no server-clock function available in a conditional write — so multi-node deployments
 112    // should keep worker clocks synchronized well inside the lease window. (DynamoDB's own TTL
 113    // reaper, by contrast, runs on the service clock against the epoch-seconds expiry attribute.)
 114
 115    private readonly IAmazonDynamoDB _client;
 116    private readonly DynamoDbDurableFlowOptions _options;
 253117    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 118    private readonly bool _ownsClient;
 119    private volatile bool _created;
 120
 253121    public DynamoDbFlowStateStore(IAmazonDynamoDB client, IOptions<DynamoDbDurableFlowOptions> options, bool ownsClient 
 122    {
 253123        _client = client;
 253124        _options = options.Value;
 253125        _options.Validate();
 253126        _ownsClient = ownsClient;
 253127    }
 128
 129    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 130    {
 726131        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 726132        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 133
 708134        var response = await _client.GetItemAsync(new GetItemRequest
 708135        {
 708136            TableName = _options.TableName,
 708137            Key = Key(flowId),
 708138            ConsistentRead = true
 708139        }, cancellationToken).ConfigureAwait(false);
 140
 141        // No item: the run is genuinely gone. This is the ONLY shape that may read as absence,
 142        // because callers acknowledge the wake-up on null.
 708143        if (response.Item is null || response.Item.Count == 0)
 15144            return null;
 145
 146        // From here the item EXISTS. A required attribute that is missing or unparseable means this
 147        // build cannot interpret a ledger that is still physically there — categorically different
 148        // from absence, and reporting it as absence acknowledged a live run's only wake-up. Only a
 149        // well-formed, elapsed TTL is real expiry.
 693150        if (!response.Item.TryGetValue(_options.TimeToLiveAttributeName, out var expires)
 693151            || !long.TryParse(expires.N, NumberStyles.Integer, CultureInfo.InvariantCulture, out var expiresAt))
 4152            throw new FlowStateUnreadableException(flowId, $"its '{_options.TimeToLiveAttributeName}' attribute is missi
 153
 689154        if (expiresAt <= DateTimeOffset.UtcNow.ToUnixTimeSeconds())
 3155            return null;
 156
 686157        if (!response.Item.TryGetValue(StateJsonAttribute, out var json) || string.IsNullOrEmpty(json.S))
 4158            throw new FlowStateUnreadableException(flowId, $"its '{StateJsonAttribute}' attribute is missing or empty");
 159
 682160        if (!response.Item.TryGetValue(RevisionAttribute, out var revision)
 682161            || !long.TryParse(revision.N, NumberStyles.Integer, CultureInfo.InvariantCulture, out var value))
 7162            throw new FlowStateUnreadableException(flowId, $"its '{RevisionAttribute}' attribute is missing or not a num
 163
 675164        return DurableFlowStoreShared.ReadState(flowId, json.S, value);
 689165    }
 166
 167    /// <inheritdoc />
 168    public void ValidateCreate(string flowId, FlowState state, TimeSpan ttl)
 169    {
 134170        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 134171        if (_options.MaxStateBytes is not null)
 134172            _ = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "DynamoDB");
 132173    }
 174
 175    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 176    {
 299177        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 298178        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "DynamoDB");
 296179        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 180
 294181        var now = DateTimeOffset.UtcNow;
 182        try
 183        {
 294184            await _client.PutItemAsync(new PutItemRequest
 294185            {
 294186                TableName = _options.TableName,
 294187                Item = CreateItem(flowId, stateJson, state.Revision, ttl, now),
 294188                ConditionExpression = "attribute_not_exists(#flow_id) OR #expires <= :now",
 294189                ExpressionAttributeNames = new Dictionary<string, string>
 294190                {
 294191                    ["#flow_id"] = FlowIdAttribute,
 294192                    ["#expires"] = _options.TimeToLiveAttributeName
 294193                },
 294194                ExpressionAttributeValues = new Dictionary<string, AttributeValue>
 294195                {
 294196                    [":now"] = new() { N = UnixSeconds(now) }
 294197                }
 294198            }, cancellationToken).ConfigureAwait(false);
 145199            return true;
 200        }
 149201        catch (ConditionalCheckFailedException)
 202        {
 149203            return false;
 204        }
 294205    }
 206
 207    public async Task<bool> TryUpdateAsync(
 208        string flowId,
 209        FlowState state,
 210        long expectedRevision,
 211        TimeSpan ttl,
 212        string? leaseId = null,
 213        CancellationToken cancellationToken = default)
 214    {
 869215        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 869216        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "DynamoDB");
 867217        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 218
 867219        var now = DateTimeOffset.UtcNow;
 867220        var names = new Dictionary<string, string>
 867221        {
 867222            ["#state"] = StateJsonAttribute,
 867223            ["#expires"] = _options.TimeToLiveAttributeName,
 867224            ["#updated"] = UpdatedAtAttribute,
 867225            ["#revision"] = RevisionAttribute
 867226        };
 867227        var values = new Dictionary<string, AttributeValue>
 867228        {
 867229            [":state"] = new() { S = stateJson },
 867230            [":expires"] = new() { N = UnixSecondsCeiling(DurableFlowStoreShared.AddSaturating(now, ttl)) },
 867231            [":updated"] = new() { N = UnixSeconds(now) },
 867232            [":expected_revision"] = new() { N = expectedRevision.ToString(CultureInfo.InvariantCulture) },
 867233            [":new_revision"] = new() { N = state.Revision.ToString(CultureInfo.InvariantCulture) },
 867234            [":now"] = new() { N = UnixSeconds(now) }
 867235        };
 867236        var condition = "#revision = :expected_revision AND #expires > :now";
 867237        if (leaseId is not null)
 238        {
 863239            condition += " AND #lease_id = :lease_id AND #lease_expires > :now_ms";
 863240            names["#lease_id"] = LeaseIdAttribute;
 863241            names["#lease_expires"] = LeaseExpiresAtAttribute;
 863242            values[":lease_id"] = new AttributeValue { S = leaseId };
 863243            values[":now_ms"] = new AttributeValue { N = UnixMilliseconds(now) };
 244        }
 245
 246        try
 247        {
 867248            await _client.UpdateItemAsync(new UpdateItemRequest
 867249            {
 867250                TableName = _options.TableName,
 867251                Key = Key(flowId),
 867252                UpdateExpression = "SET #state = :state, #expires = :expires, #updated = :updated, #revision = :new_revi
 867253                ConditionExpression = condition,
 867254                ExpressionAttributeNames = names,
 867255                ExpressionAttributeValues = values
 867256            }, cancellationToken).ConfigureAwait(false);
 862257            return true;
 258        }
 5259        catch (ConditionalCheckFailedException)
 260        {
 5261            return false;
 262        }
 867263    }
 264
 265    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 157266        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 267
 268    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 11269        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 270
 271    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 272    {
 146273        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 274        try
 275        {
 146276            await _client.UpdateItemAsync(new UpdateItemRequest
 146277            {
 146278                TableName = _options.TableName,
 146279                Key = Key(flowId),
 146280                UpdateExpression = "REMOVE #lease_id, #lease_expires",
 146281                ConditionExpression = "#lease_id = :lease_id",
 146282                ExpressionAttributeNames = new Dictionary<string, string>
 146283                {
 146284                    ["#lease_id"] = LeaseIdAttribute,
 146285                    ["#lease_expires"] = LeaseExpiresAtAttribute
 146286                },
 146287                ExpressionAttributeValues = new Dictionary<string, AttributeValue>
 146288                {
 146289                    [":lease_id"] = new() { S = leaseId }
 146290                }
 146291            }, cancellationToken).ConfigureAwait(false);
 141292        }
 5293        catch (ConditionalCheckFailedException)
 294        {
 5295        }
 146296    }
 297
 298    /// <inheritdoc />
 299    public async Task<FlowLeaseObservation?> ObserveLeaseAsync(string flowId, CancellationToken cancellationToken = defa
 300    {
 44301        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 38302        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 303
 304        // The two lease attributes exactly as stored — deliberately never compared with a clock,
 305        // unlike every other operation in this store: an expired lease nobody has taken over must
 306        // keep reading as the same lease, because the engine's proof of a live holder is that two
 307        // observations DIFFER. Whether it has lapsed stays UpdateLeaseAsync's call. Consistent
 308        // for the reason LoadAsync is — an eventually consistent read can replay a lease a renewal
 309        // has already moved, and "unchanged" is the one answer that must never be stale — and
 310        // projected, so state_json (up to the 400 KB item cap) stays off the wire.
 38311        var response = await _client.GetItemAsync(new GetItemRequest
 38312        {
 38313            TableName = _options.TableName,
 38314            Key = Key(flowId),
 38315            ConsistentRead = true,
 38316            ProjectionExpression = "#lease_id, #lease_expires",
 38317            ExpressionAttributeNames = new Dictionary<string, string>
 38318            {
 38319                ["#lease_id"] = LeaseIdAttribute,
 38320                ["#lease_expires"] = LeaseExpiresAtAttribute
 38321            }
 38322        }, cancellationToken).ConfigureAwait(false);
 323
 324        // No item and an item with neither attribute both come back empty under a projection;
 325        // either way nobody holds the lease.
 38326        if (response.Item is null
 38327            || !response.Item.TryGetValue(LeaseIdAttribute, out var leaseId)
 38328            || string.IsNullOrEmpty(leaseId.S))
 14329            return FlowLeaseObservation.Unheld;
 330
 331        // lease_expires_at_ms is epoch milliseconds (UpdateLeaseAsync writes it that way). A holder
 332        // whose expiry is missing or unreadable is still a holder: the owner is reported and the
 333        // expiry left null, which the engine reads as "no persisted deadline to wait out".
 24334        return DurableFlowStoreShared.LeaseObservation(
 24335            leaseId.S,
 24336            response.Item.TryGetValue(LeaseExpiresAtAttribute, out var leaseExpires)
 24337            && long.TryParse(leaseExpires.N, NumberStyles.Integer, CultureInfo.InvariantCulture, out var expiresAtMs)
 24338            && expiresAtMs >= MinUnixMilliseconds
 24339            && expiresAtMs <= MaxUnixMilliseconds
 24340                ? DateTimeOffset.FromUnixTimeMilliseconds(expiresAtMs).UtcDateTime
 24341                : null);
 38342    }
 343
 344    // The range DateTimeOffset.FromUnixTimeMilliseconds accepts; a hand-edited attribute outside
 345    // it would otherwise throw ArgumentOutOfRangeException out of an observation.
 3346    private static readonly long MinUnixMilliseconds = DateTimeOffset.MinValue.ToUnixTimeMilliseconds();
 3347    private static readonly long MaxUnixMilliseconds = DateTimeOffset.MaxValue.ToUnixTimeMilliseconds();
 348
 349    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 350    {
 14351        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 14352        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 353
 14354        var response = await _client.DeleteItemAsync(new DeleteItemRequest
 14355        {
 14356            TableName = _options.TableName,
 14357            Key = Key(flowId),
 14358            ReturnValues = ReturnValue.ALL_OLD
 14359        }, cancellationToken).ConfigureAwait(false);
 14360        return response.Attributes is { Count: > 0 };
 14361    }
 362
 363    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 364    {
 2253365        if (_created)
 1956366            return;
 367
 297368        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 369        try
 370        {
 297371            if (_created)
 116372                return;
 373
 181374            TableDescription? table = null;
 375            try
 376            {
 181377                var described = await _client.DescribeTableAsync(_options.TableName, cancellationToken).ConfigureAwait(f
 42378                table = described.Table;
 42379            }
 139380            catch (ResourceNotFoundException)
 381            {
 139382            }
 383
 181384            if (table is null)
 385            {
 139386                if (!_options.AutoCreateTable)
 2387                    throw new InvalidOperationException(
 2388                        $"DynamoDB table '{_options.TableName}' does not exist and {nameof(DynamoDbDurableFlowOptions.Au
 389
 390                try
 391                {
 137392                    await _client.CreateTableAsync(new CreateTableRequest
 137393                    {
 137394                        TableName = _options.TableName,
 137395                        BillingMode = BillingMode.PAY_PER_REQUEST,
 137396                        AttributeDefinitions =
 137397                        [
 137398                            new AttributeDefinition(FlowIdAttribute, ScalarAttributeType.S)
 137399                        ],
 137400                        KeySchema =
 137401                        [
 137402                            new KeySchemaElement(FlowIdAttribute, KeyType.HASH)
 137403                        ]
 137404                    }, cancellationToken).ConfigureAwait(false);
 135405                }
 2406                catch (ResourceInUseException)
 407                {
 408                    // Another process won the create race; fall through and wait for ACTIVE.
 2409                }
 410
 137411                table = await WaitForTableActiveAsync(cancellationToken).ConfigureAwait(false);
 412            }
 42413            else if (table.TableStatus != TableStatus.ACTIVE)
 414            {
 2415                table = await WaitForTableActiveAsync(cancellationToken).ConfigureAwait(false);
 416            }
 417
 179418            ValidateTableSchema(table);
 419
 420            // Verification runs for every operator-provisioned table, not only when
 421            // EnableTimeToLive is set: the flag governs ENABLING (an auto-create concern, as its
 422            // doc says), but TTL is this store's only cleanup mechanism (no application-side
 423            // pruning exists), so an operator who provisioned the table and turned the flag off
 424            // was thereby turning off the check that their table actually has TTL — and it grew
 425            // without bound with no error and no log line.
 171426            if (_options.EnableTimeToLive || !_options.AutoCreateTable)
 427            {
 428                // Check the TTL status instead of blind-enabling: UpdateTimeToLive throws when TTL
 429                // is already enabled, and relying on a swallowed exception per provisioning is noise.
 153430                var ttlStatus = await _client.DescribeTimeToLiveAsync(new DescribeTimeToLiveRequest
 153431                {
 153432                    TableName = _options.TableName
 153433                }, cancellationToken).ConfigureAwait(false);
 434
 153435                var description = ttlStatus.TimeToLiveDescription;
 153436                var status = description?.TimeToLiveStatus;
 153437                if ((status == TimeToLiveStatus.ENABLED || status == TimeToLiveStatus.ENABLING)
 153438                    && !string.Equals(description?.AttributeName, _options.TimeToLiveAttributeName, StringComparison.Ord
 439                {
 4440                    throw new InvalidOperationException(
 4441                        $"DynamoDB table '{_options.TableName}' has TTL configured on attribute '{description?.Attribute
 4442                        $"but '{_options.TimeToLiveAttributeName}' is required.");
 443                }
 444
 149445                if (status != TimeToLiveStatus.ENABLED && status != TimeToLiveStatus.ENABLING)
 446                {
 79447                    if (!_options.AutoCreateTable)
 448                    {
 4449                        throw new InvalidOperationException(
 4450                            $"DynamoDB table '{_options.TableName}' does not have TTL enabled on " +
 4451                            $"'{_options.TimeToLiveAttributeName}'. Enable it in infrastructure before using the table."
 452                    }
 453
 454                    try
 455                    {
 75456                        await _client.UpdateTimeToLiveAsync(new UpdateTimeToLiveRequest
 75457                        {
 75458                            TableName = _options.TableName,
 75459                            TimeToLiveSpecification = new TimeToLiveSpecification
 75460                            {
 75461                                AttributeName = _options.TimeToLiveAttributeName,
 75462                                Enabled = true
 75463                            }
 75464                        }, cancellationToken).ConfigureAwait(false);
 71465                    }
 4466                    catch (AmazonDynamoDBException ex) when (string.Equals(ex.ErrorCode, "ValidationException", StringCo
 467                    {
 468                        // Accept only the one safe race: another process enabled the expected TTL
 469                        // attribute between our describe and update. Do not swallow unrelated
 470                        // validation failures.
 4471                        var raced = await _client.DescribeTimeToLiveAsync(new DescribeTimeToLiveRequest
 4472                        {
 4473                            TableName = _options.TableName
 4474                        }, cancellationToken).ConfigureAwait(false);
 4475                        var racedDescription = raced.TimeToLiveDescription;
 4476                        if ((racedDescription?.TimeToLiveStatus != TimeToLiveStatus.ENABLED
 4477                                && racedDescription?.TimeToLiveStatus != TimeToLiveStatus.ENABLING)
 4478                            || !string.Equals(racedDescription.AttributeName, _options.TimeToLiveAttributeName, StringCo
 479                        {
 2480                            throw;
 481                        }
 482                    }
 483                }
 484            }
 485
 161486            _created = true;
 161487        }
 488        finally
 489        {
 297490            _ensureGate.Release();
 491        }
 2233492    }
 493
 494    private async Task<TableDescription> WaitForTableActiveAsync(CancellationToken cancellationToken)
 495    {
 139496        var deadline = DateTime.UtcNow.AddSeconds(30);
 2497        while (true)
 498        {
 141499            var response = await _client.DescribeTableAsync(_options.TableName, cancellationToken).ConfigureAwait(false)
 141500            if (response.Table.TableStatus == TableStatus.ACTIVE)
 139501                return response.Table;
 2502            if (DateTime.UtcNow >= deadline)
 0503                throw new TimeoutException($"DynamoDB table '{_options.TableName}' did not become ACTIVE within 30 secon
 504
 2505            await Task.Delay(TimeSpan.FromMilliseconds(250), cancellationToken).ConfigureAwait(false);
 506        }
 139507    }
 508
 509    private void ValidateTableSchema(TableDescription table)
 510    {
 356511        var hashKey = table.KeySchema?.SingleOrDefault(key => key.KeyType == KeyType.HASH);
 179512        var keyDefinition = table.AttributeDefinitions?
 356513            .SingleOrDefault(attribute => string.Equals(attribute.AttributeName, FlowIdAttribute, StringComparison.Ordin
 179514        if (table.KeySchema?.Count != 1
 179515            || !string.Equals(hashKey?.AttributeName, FlowIdAttribute, StringComparison.Ordinal)
 179516            || keyDefinition?.AttributeType != ScalarAttributeType.S)
 517        {
 8518            throw new InvalidOperationException(
 8519                $"DynamoDB table '{_options.TableName}' must use one string partition key named '{FlowIdAttribute}' and 
 520        }
 171521    }
 522
 523    private async Task<bool> UpdateLeaseAsync(
 524        string flowId,
 525        string leaseId,
 526        TimeSpan leaseDuration,
 527        bool acquire,
 528        CancellationToken cancellationToken)
 529    {
 168530        DurableFlowStoreShared.ValidateLeaseArgs(flowId, leaseId, leaseDuration);
 531
 166532        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 166533        var now = DateTimeOffset.UtcNow;
 534        try
 535        {
 166536            await _client.UpdateItemAsync(new UpdateItemRequest
 166537            {
 166538                TableName = _options.TableName,
 166539                Key = Key(flowId),
 166540                UpdateExpression = "SET #lease_id = :lease_id, #lease_expires = :lease_expires",
 166541                ConditionExpression = acquire
 166542                    ? "#expires > :now AND attribute_exists(#revision) AND (attribute_not_exists(#lease_id) OR #lease_ex
 166543                    : "#expires > :now AND attribute_exists(#revision) AND #lease_id = :lease_id AND #lease_expires > :n
 166544                ExpressionAttributeNames = new Dictionary<string, string>
 166545                {
 166546                    ["#expires"] = _options.TimeToLiveAttributeName,
 166547                    ["#revision"] = RevisionAttribute,
 166548                    ["#lease_id"] = LeaseIdAttribute,
 166549                    ["#lease_expires"] = LeaseExpiresAtAttribute
 166550                },
 166551                ExpressionAttributeValues = new Dictionary<string, AttributeValue>
 166552                {
 166553                    [":now"] = new() { N = UnixSeconds(now) },
 166554                    [":now_ms"] = new() { N = UnixMilliseconds(now) },
 166555                    [":lease_id"] = new() { S = leaseId },
 166556                    [":lease_expires"] = new() { N = UnixMilliseconds(DurableFlowStoreShared.AddSaturating(now, leaseDur
 166557                }
 166558            }, cancellationToken).ConfigureAwait(false);
 152559            return true;
 560        }
 14561        catch (ConditionalCheckFailedException)
 562        {
 14563            return false;
 564        }
 166565    }
 566
 567    private Dictionary<string, AttributeValue> CreateItem(string flowId, string stateJson, long revision, TimeSpan ttl, 
 294568        => new(StringComparer.Ordinal)
 294569        {
 294570            [FlowIdAttribute] = new() { S = flowId },
 294571            [StateJsonAttribute] = new() { S = stateJson },
 294572            [_options.TimeToLiveAttributeName] = new() { N = UnixSecondsCeiling(DurableFlowStoreShared.AddSaturating(now
 294573            [UpdatedAtAttribute] = new() { N = UnixSeconds(now) },
 294574            [RevisionAttribute] = new() { N = revision.ToString(CultureInfo.InvariantCulture) }
 294575        };
 576
 577    private static Dictionary<string, AttributeValue> Key(string flowId)
 1939578        => new(StringComparer.Ordinal) { [FlowIdAttribute] = new AttributeValue { S = flowId } };
 579
 580    private static string UnixSeconds(DateTimeOffset value)
 2488581        => value.ToUnixTimeSeconds().ToString(CultureInfo.InvariantCulture);
 582
 583    private static string UnixSecondsCeiling(DateTimeOffset value)
 1161584        => ((long)Math.Ceiling(value.ToUnixTimeMilliseconds() / 1000.0)).ToString(CultureInfo.InvariantCulture);
 585
 586    private static string UnixMilliseconds(DateTimeOffset value)
 1195587        => value.ToUnixTimeMilliseconds().ToString(CultureInfo.InvariantCulture);
 588
 589    /// <summary>Disposes the DynamoDB client when the store created (and therefore owns) it.</summary>
 590    public void Dispose()
 591    {
 446592        _ensureGate.Dispose();
 446593        if (_ownsClient)
 2594            _client.Dispose();
 446595    }
 596}
 597}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(Amazon.DynamoDBv2.IAmazonDynamoDB,Microsoft.Extensions.Options.IOptions`1<AsyncResponse.DurableFlows.DynamoDB.DynamoDbDurableFlowOptions>,System.Boolean)
LoadAsync()
ValidateCreate(System.String,AsyncResponse.FlowState,System.TimeSpan)
TryCreateAsync()
TryUpdateAsync()
TryAcquireLeaseAsync(System.String,System.String,System.TimeSpan,System.Threading.CancellationToken)
TryRenewLeaseAsync(System.String,System.String,System.TimeSpan,System.Threading.CancellationToken)
ReleaseLeaseAsync()
ObserveLeaseAsync()
.cctor()
TryDeleteAsync()
EnsureCreatedAsync()
WaitForTableActiveAsync()
ValidateTableSchema(Amazon.DynamoDBv2.Model.TableDescription)
UpdateLeaseAsync()
CreateItem(System.String,System.String,System.Int64,System.TimeSpan,System.DateTimeOffset)
Key(System.String)
UnixSeconds(System.DateTimeOffset)
UnixSecondsCeiling(System.DateTimeOffset)
UnixMilliseconds(System.DateTimeOffset)
Dispose()