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

Information
Class: Microsoft.Extensions.DependencyInjection.DynamoDbDurableFlowServiceCollectionExtensions
Assembly: AsyncResponse.DurableFlows.DynamoDB
File(s): /_/src/DurableFlows/AsyncResponse.DurableFlows.DynamoDB/DynamoDbDurableFlows.cs
Line coverage
100%
Covered lines: 9
Uncovered lines: 0
Coverable lines: 9
Total lines: 597
Line coverage: 100%
Branch coverage
50%
Covered branches: 1
Total branches: 2
Branch coverage: 50%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
WithDynamoDbDurableFlows(...)50%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.
 20029            builder.Services.TryAddSingleton(provider =>
 20030            {
 20031                var options = provider.GetRequiredService<IOptions<DynamoDbDurableFlowOptions>>();
 20032                var shared = provider.GetService<IAmazonDynamoDB>();
 20033                return shared is not null
 20034                    ? new DynamoDbFlowStateStore(shared, options)
 20035                    : new DynamoDbFlowStateStore(new AmazonDynamoDBClient(), options, ownsClient: true);
 20036            });
 20037            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;
 117    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 118    private readonly bool _ownsClient;
 119    private volatile bool _created;
 120
 121    public DynamoDbFlowStateStore(IAmazonDynamoDB client, IOptions<DynamoDbDurableFlowOptions> options, bool ownsClient 
 122    {
 123        _client = client;
 124        _options = options.Value;
 125        _options.Validate();
 126        _ownsClient = ownsClient;
 127    }
 128
 129    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 130    {
 131        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 132        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 133
 134        var response = await _client.GetItemAsync(new GetItemRequest
 135        {
 136            TableName = _options.TableName,
 137            Key = Key(flowId),
 138            ConsistentRead = true
 139        }, 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.
 143        if (response.Item is null || response.Item.Count == 0)
 144            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.
 150        if (!response.Item.TryGetValue(_options.TimeToLiveAttributeName, out var expires)
 151            || !long.TryParse(expires.N, NumberStyles.Integer, CultureInfo.InvariantCulture, out var expiresAt))
 152            throw new FlowStateUnreadableException(flowId, $"its '{_options.TimeToLiveAttributeName}' attribute is missi
 153
 154        if (expiresAt <= DateTimeOffset.UtcNow.ToUnixTimeSeconds())
 155            return null;
 156
 157        if (!response.Item.TryGetValue(StateJsonAttribute, out var json) || string.IsNullOrEmpty(json.S))
 158            throw new FlowStateUnreadableException(flowId, $"its '{StateJsonAttribute}' attribute is missing or empty");
 159
 160        if (!response.Item.TryGetValue(RevisionAttribute, out var revision)
 161            || !long.TryParse(revision.N, NumberStyles.Integer, CultureInfo.InvariantCulture, out var value))
 162            throw new FlowStateUnreadableException(flowId, $"its '{RevisionAttribute}' attribute is missing or not a num
 163
 164        return DurableFlowStoreShared.ReadState(flowId, json.S, value);
 165    }
 166
 167    /// <inheritdoc />
 168    public void ValidateCreate(string flowId, FlowState state, TimeSpan ttl)
 169    {
 170        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 171        if (_options.MaxStateBytes is not null)
 172            _ = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "DynamoDB");
 173    }
 174
 175    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 176    {
 177        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 178        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "DynamoDB");
 179        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 180
 181        var now = DateTimeOffset.UtcNow;
 182        try
 183        {
 184            await _client.PutItemAsync(new PutItemRequest
 185            {
 186                TableName = _options.TableName,
 187                Item = CreateItem(flowId, stateJson, state.Revision, ttl, now),
 188                ConditionExpression = "attribute_not_exists(#flow_id) OR #expires <= :now",
 189                ExpressionAttributeNames = new Dictionary<string, string>
 190                {
 191                    ["#flow_id"] = FlowIdAttribute,
 192                    ["#expires"] = _options.TimeToLiveAttributeName
 193                },
 194                ExpressionAttributeValues = new Dictionary<string, AttributeValue>
 195                {
 196                    [":now"] = new() { N = UnixSeconds(now) }
 197                }
 198            }, cancellationToken).ConfigureAwait(false);
 199            return true;
 200        }
 201        catch (ConditionalCheckFailedException)
 202        {
 203            return false;
 204        }
 205    }
 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    {
 215        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 216        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "DynamoDB");
 217        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 218
 219        var now = DateTimeOffset.UtcNow;
 220        var names = new Dictionary<string, string>
 221        {
 222            ["#state"] = StateJsonAttribute,
 223            ["#expires"] = _options.TimeToLiveAttributeName,
 224            ["#updated"] = UpdatedAtAttribute,
 225            ["#revision"] = RevisionAttribute
 226        };
 227        var values = new Dictionary<string, AttributeValue>
 228        {
 229            [":state"] = new() { S = stateJson },
 230            [":expires"] = new() { N = UnixSecondsCeiling(DurableFlowStoreShared.AddSaturating(now, ttl)) },
 231            [":updated"] = new() { N = UnixSeconds(now) },
 232            [":expected_revision"] = new() { N = expectedRevision.ToString(CultureInfo.InvariantCulture) },
 233            [":new_revision"] = new() { N = state.Revision.ToString(CultureInfo.InvariantCulture) },
 234            [":now"] = new() { N = UnixSeconds(now) }
 235        };
 236        var condition = "#revision = :expected_revision AND #expires > :now";
 237        if (leaseId is not null)
 238        {
 239            condition += " AND #lease_id = :lease_id AND #lease_expires > :now_ms";
 240            names["#lease_id"] = LeaseIdAttribute;
 241            names["#lease_expires"] = LeaseExpiresAtAttribute;
 242            values[":lease_id"] = new AttributeValue { S = leaseId };
 243            values[":now_ms"] = new AttributeValue { N = UnixMilliseconds(now) };
 244        }
 245
 246        try
 247        {
 248            await _client.UpdateItemAsync(new UpdateItemRequest
 249            {
 250                TableName = _options.TableName,
 251                Key = Key(flowId),
 252                UpdateExpression = "SET #state = :state, #expires = :expires, #updated = :updated, #revision = :new_revi
 253                ConditionExpression = condition,
 254                ExpressionAttributeNames = names,
 255                ExpressionAttributeValues = values
 256            }, cancellationToken).ConfigureAwait(false);
 257            return true;
 258        }
 259        catch (ConditionalCheckFailedException)
 260        {
 261            return false;
 262        }
 263    }
 264
 265    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 266        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 267
 268    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 269        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 270
 271    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 272    {
 273        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 274        try
 275        {
 276            await _client.UpdateItemAsync(new UpdateItemRequest
 277            {
 278                TableName = _options.TableName,
 279                Key = Key(flowId),
 280                UpdateExpression = "REMOVE #lease_id, #lease_expires",
 281                ConditionExpression = "#lease_id = :lease_id",
 282                ExpressionAttributeNames = new Dictionary<string, string>
 283                {
 284                    ["#lease_id"] = LeaseIdAttribute,
 285                    ["#lease_expires"] = LeaseExpiresAtAttribute
 286                },
 287                ExpressionAttributeValues = new Dictionary<string, AttributeValue>
 288                {
 289                    [":lease_id"] = new() { S = leaseId }
 290                }
 291            }, cancellationToken).ConfigureAwait(false);
 292        }
 293        catch (ConditionalCheckFailedException)
 294        {
 295        }
 296    }
 297
 298    /// <inheritdoc />
 299    public async Task<FlowLeaseObservation?> ObserveLeaseAsync(string flowId, CancellationToken cancellationToken = defa
 300    {
 301        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 302        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.
 311        var response = await _client.GetItemAsync(new GetItemRequest
 312        {
 313            TableName = _options.TableName,
 314            Key = Key(flowId),
 315            ConsistentRead = true,
 316            ProjectionExpression = "#lease_id, #lease_expires",
 317            ExpressionAttributeNames = new Dictionary<string, string>
 318            {
 319                ["#lease_id"] = LeaseIdAttribute,
 320                ["#lease_expires"] = LeaseExpiresAtAttribute
 321            }
 322        }, 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.
 326        if (response.Item is null
 327            || !response.Item.TryGetValue(LeaseIdAttribute, out var leaseId)
 328            || string.IsNullOrEmpty(leaseId.S))
 329            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".
 334        return DurableFlowStoreShared.LeaseObservation(
 335            leaseId.S,
 336            response.Item.TryGetValue(LeaseExpiresAtAttribute, out var leaseExpires)
 337            && long.TryParse(leaseExpires.N, NumberStyles.Integer, CultureInfo.InvariantCulture, out var expiresAtMs)
 338            && expiresAtMs >= MinUnixMilliseconds
 339            && expiresAtMs <= MaxUnixMilliseconds
 340                ? DateTimeOffset.FromUnixTimeMilliseconds(expiresAtMs).UtcDateTime
 341                : null);
 342    }
 343
 344    // The range DateTimeOffset.FromUnixTimeMilliseconds accepts; a hand-edited attribute outside
 345    // it would otherwise throw ArgumentOutOfRangeException out of an observation.
 346    private static readonly long MinUnixMilliseconds = DateTimeOffset.MinValue.ToUnixTimeMilliseconds();
 347    private static readonly long MaxUnixMilliseconds = DateTimeOffset.MaxValue.ToUnixTimeMilliseconds();
 348
 349    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 350    {
 351        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 352        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 353
 354        var response = await _client.DeleteItemAsync(new DeleteItemRequest
 355        {
 356            TableName = _options.TableName,
 357            Key = Key(flowId),
 358            ReturnValues = ReturnValue.ALL_OLD
 359        }, cancellationToken).ConfigureAwait(false);
 360        return response.Attributes is { Count: > 0 };
 361    }
 362
 363    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 364    {
 365        if (_created)
 366            return;
 367
 368        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 369        try
 370        {
 371            if (_created)
 372                return;
 373
 374            TableDescription? table = null;
 375            try
 376            {
 377                var described = await _client.DescribeTableAsync(_options.TableName, cancellationToken).ConfigureAwait(f
 378                table = described.Table;
 379            }
 380            catch (ResourceNotFoundException)
 381            {
 382            }
 383
 384            if (table is null)
 385            {
 386                if (!_options.AutoCreateTable)
 387                    throw new InvalidOperationException(
 388                        $"DynamoDB table '{_options.TableName}' does not exist and {nameof(DynamoDbDurableFlowOptions.Au
 389
 390                try
 391                {
 392                    await _client.CreateTableAsync(new CreateTableRequest
 393                    {
 394                        TableName = _options.TableName,
 395                        BillingMode = BillingMode.PAY_PER_REQUEST,
 396                        AttributeDefinitions =
 397                        [
 398                            new AttributeDefinition(FlowIdAttribute, ScalarAttributeType.S)
 399                        ],
 400                        KeySchema =
 401                        [
 402                            new KeySchemaElement(FlowIdAttribute, KeyType.HASH)
 403                        ]
 404                    }, cancellationToken).ConfigureAwait(false);
 405                }
 406                catch (ResourceInUseException)
 407                {
 408                    // Another process won the create race; fall through and wait for ACTIVE.
 409                }
 410
 411                table = await WaitForTableActiveAsync(cancellationToken).ConfigureAwait(false);
 412            }
 413            else if (table.TableStatus != TableStatus.ACTIVE)
 414            {
 415                table = await WaitForTableActiveAsync(cancellationToken).ConfigureAwait(false);
 416            }
 417
 418            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.
 426            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.
 430                var ttlStatus = await _client.DescribeTimeToLiveAsync(new DescribeTimeToLiveRequest
 431                {
 432                    TableName = _options.TableName
 433                }, cancellationToken).ConfigureAwait(false);
 434
 435                var description = ttlStatus.TimeToLiveDescription;
 436                var status = description?.TimeToLiveStatus;
 437                if ((status == TimeToLiveStatus.ENABLED || status == TimeToLiveStatus.ENABLING)
 438                    && !string.Equals(description?.AttributeName, _options.TimeToLiveAttributeName, StringComparison.Ord
 439                {
 440                    throw new InvalidOperationException(
 441                        $"DynamoDB table '{_options.TableName}' has TTL configured on attribute '{description?.Attribute
 442                        $"but '{_options.TimeToLiveAttributeName}' is required.");
 443                }
 444
 445                if (status != TimeToLiveStatus.ENABLED && status != TimeToLiveStatus.ENABLING)
 446                {
 447                    if (!_options.AutoCreateTable)
 448                    {
 449                        throw new InvalidOperationException(
 450                            $"DynamoDB table '{_options.TableName}' does not have TTL enabled on " +
 451                            $"'{_options.TimeToLiveAttributeName}'. Enable it in infrastructure before using the table."
 452                    }
 453
 454                    try
 455                    {
 456                        await _client.UpdateTimeToLiveAsync(new UpdateTimeToLiveRequest
 457                        {
 458                            TableName = _options.TableName,
 459                            TimeToLiveSpecification = new TimeToLiveSpecification
 460                            {
 461                                AttributeName = _options.TimeToLiveAttributeName,
 462                                Enabled = true
 463                            }
 464                        }, cancellationToken).ConfigureAwait(false);
 465                    }
 466                    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.
 471                        var raced = await _client.DescribeTimeToLiveAsync(new DescribeTimeToLiveRequest
 472                        {
 473                            TableName = _options.TableName
 474                        }, cancellationToken).ConfigureAwait(false);
 475                        var racedDescription = raced.TimeToLiveDescription;
 476                        if ((racedDescription?.TimeToLiveStatus != TimeToLiveStatus.ENABLED
 477                                && racedDescription?.TimeToLiveStatus != TimeToLiveStatus.ENABLING)
 478                            || !string.Equals(racedDescription.AttributeName, _options.TimeToLiveAttributeName, StringCo
 479                        {
 480                            throw;
 481                        }
 482                    }
 483                }
 484            }
 485
 486            _created = true;
 487        }
 488        finally
 489        {
 490            _ensureGate.Release();
 491        }
 492    }
 493
 494    private async Task<TableDescription> WaitForTableActiveAsync(CancellationToken cancellationToken)
 495    {
 496        var deadline = DateTime.UtcNow.AddSeconds(30);
 497        while (true)
 498        {
 499            var response = await _client.DescribeTableAsync(_options.TableName, cancellationToken).ConfigureAwait(false)
 500            if (response.Table.TableStatus == TableStatus.ACTIVE)
 501                return response.Table;
 502            if (DateTime.UtcNow >= deadline)
 503                throw new TimeoutException($"DynamoDB table '{_options.TableName}' did not become ACTIVE within 30 secon
 504
 505            await Task.Delay(TimeSpan.FromMilliseconds(250), cancellationToken).ConfigureAwait(false);
 506        }
 507    }
 508
 509    private void ValidateTableSchema(TableDescription table)
 510    {
 511        var hashKey = table.KeySchema?.SingleOrDefault(key => key.KeyType == KeyType.HASH);
 512        var keyDefinition = table.AttributeDefinitions?
 513            .SingleOrDefault(attribute => string.Equals(attribute.AttributeName, FlowIdAttribute, StringComparison.Ordin
 514        if (table.KeySchema?.Count != 1
 515            || !string.Equals(hashKey?.AttributeName, FlowIdAttribute, StringComparison.Ordinal)
 516            || keyDefinition?.AttributeType != ScalarAttributeType.S)
 517        {
 518            throw new InvalidOperationException(
 519                $"DynamoDB table '{_options.TableName}' must use one string partition key named '{FlowIdAttribute}' and 
 520        }
 521    }
 522
 523    private async Task<bool> UpdateLeaseAsync(
 524        string flowId,
 525        string leaseId,
 526        TimeSpan leaseDuration,
 527        bool acquire,
 528        CancellationToken cancellationToken)
 529    {
 530        DurableFlowStoreShared.ValidateLeaseArgs(flowId, leaseId, leaseDuration);
 531
 532        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 533        var now = DateTimeOffset.UtcNow;
 534        try
 535        {
 536            await _client.UpdateItemAsync(new UpdateItemRequest
 537            {
 538                TableName = _options.TableName,
 539                Key = Key(flowId),
 540                UpdateExpression = "SET #lease_id = :lease_id, #lease_expires = :lease_expires",
 541                ConditionExpression = acquire
 542                    ? "#expires > :now AND attribute_exists(#revision) AND (attribute_not_exists(#lease_id) OR #lease_ex
 543                    : "#expires > :now AND attribute_exists(#revision) AND #lease_id = :lease_id AND #lease_expires > :n
 544                ExpressionAttributeNames = new Dictionary<string, string>
 545                {
 546                    ["#expires"] = _options.TimeToLiveAttributeName,
 547                    ["#revision"] = RevisionAttribute,
 548                    ["#lease_id"] = LeaseIdAttribute,
 549                    ["#lease_expires"] = LeaseExpiresAtAttribute
 550                },
 551                ExpressionAttributeValues = new Dictionary<string, AttributeValue>
 552                {
 553                    [":now"] = new() { N = UnixSeconds(now) },
 554                    [":now_ms"] = new() { N = UnixMilliseconds(now) },
 555                    [":lease_id"] = new() { S = leaseId },
 556                    [":lease_expires"] = new() { N = UnixMilliseconds(DurableFlowStoreShared.AddSaturating(now, leaseDur
 557                }
 558            }, cancellationToken).ConfigureAwait(false);
 559            return true;
 560        }
 561        catch (ConditionalCheckFailedException)
 562        {
 563            return false;
 564        }
 565    }
 566
 567    private Dictionary<string, AttributeValue> CreateItem(string flowId, string stateJson, long revision, TimeSpan ttl, 
 568        => new(StringComparer.Ordinal)
 569        {
 570            [FlowIdAttribute] = new() { S = flowId },
 571            [StateJsonAttribute] = new() { S = stateJson },
 572            [_options.TimeToLiveAttributeName] = new() { N = UnixSecondsCeiling(DurableFlowStoreShared.AddSaturating(now
 573            [UpdatedAtAttribute] = new() { N = UnixSeconds(now) },
 574            [RevisionAttribute] = new() { N = revision.ToString(CultureInfo.InvariantCulture) }
 575        };
 576
 577    private static Dictionary<string, AttributeValue> Key(string flowId)
 578        => new(StringComparer.Ordinal) { [FlowIdAttribute] = new AttributeValue { S = flowId } };
 579
 580    private static string UnixSeconds(DateTimeOffset value)
 581        => value.ToUnixTimeSeconds().ToString(CultureInfo.InvariantCulture);
 582
 583    private static string UnixSecondsCeiling(DateTimeOffset value)
 584        => ((long)Math.Ceiling(value.ToUnixTimeMilliseconds() / 1000.0)).ToString(CultureInfo.InvariantCulture);
 585
 586    private static string UnixMilliseconds(DateTimeOffset value)
 587        => 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    {
 592        _ensureGate.Dispose();
 593        if (_ownsClient)
 594            _client.Dispose();
 595    }
 596}
 597}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

WithDynamoDbDurableFlows(Microsoft.Extensions.DependencyInjection.AsyncResponseRegistrationBuilder,System.Action`1<AsyncResponse.DurableFlows.DynamoDB.DynamoDbDurableFlowOptions>)