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

Information
Class: Microsoft.Extensions.DependencyInjection.CosmosDurableFlowServiceCollectionExtensions
Assembly: AsyncResponse.DurableFlows.Cosmos
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.Cosmos/CosmosDurableFlows.cs
Line coverage
100%
Covered lines: 13
Uncovered lines: 0
Coverable lines: 13
Total lines: 573
Line coverage: 100%
Branch coverage
100%
Covered branches: 6
Total branches: 6
Branch coverage: 100%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
WithCosmosDurableFlows(...)100%22100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.Cosmos/CosmosDurableFlows.cs

#LineLine coverage
 1using AsyncResponse;
 2using AsyncResponse.DurableFlows.Internal;
 3using AsyncResponse.DurableFlows.Cosmos;
 4using Microsoft.Azure.Cosmos;
 5using Microsoft.Extensions.DependencyInjection.Extensions;
 6using Microsoft.Extensions.Options;
 7using Newtonsoft.Json;
 8using System.Net;
 9
 10namespace Microsoft.Extensions.DependencyInjection
 11{
 12    /// <summary>DI registration for the Azure Cosmos DB durable-flow state store.</summary>
 13    public static class CosmosDurableFlowServiceCollectionExtensions
 14    {
 15        /// <summary>
 16        /// Stores durable-flow state in Azure Cosmos DB. Hosts may either register a
 17        /// <see cref="CosmosClient"/> singleton or set connection options here.
 18        /// </summary>
 19        public static AsyncResponseRegistrationBuilder WithCosmosDurableFlows(
 20            this AsyncResponseRegistrationBuilder builder,
 21            Action<CosmosDurableFlowOptions>? configure = null)
 22        {
 23            // Singleton on purpose: database/container provisioning is cached per store instance
 24            // and Cosmos metadata operations are RU-charged and rate-limited — a scoped store would
 25            // re-issue them on every flow execution. A host-registered CosmosClient is reused when
 26            // present; otherwise the store creates and owns one from ConnectionString. Nothing is
 27            // registered as a bare CosmosClient service, so unrelated resolutions of that type are
 28            // never answered — or broken — by this package.
 229            builder.Services.TryAddSingleton(provider =>
 230            {
 231                var options = provider.GetRequiredService<IOptions<CosmosDurableFlowOptions>>();
 232
 233                var shared = provider.GetService<CosmosClient>();
 234                if (shared is not null)
 235                    return new CosmosFlowStateStore(shared, options);
 236
 237                if (string.IsNullOrWhiteSpace(options.Value.ConnectionString))
 238                    throw new InvalidOperationException($"{nameof(CosmosDurableFlowOptions)}.{nameof(CosmosDurableFlowOp
 239                return new CosmosFlowStateStore(new CosmosClient(options.Value.ConnectionString), options, ownsClient: t
 240            });
 241            return builder.WithDurableFlows<CosmosFlowStateStore, CosmosDurableFlowOptions>(configure);
 42        }
 43    }
 44}
 45
 46namespace AsyncResponse.DurableFlows.Cosmos
 47{
 48/// <summary>Options for the Azure Cosmos DB durable-flow state store.</summary>
 49public sealed class CosmosDurableFlowOptions : DurableFlowOptions
 50{
 51    /// <summary>Optional Cosmos DB connection string used when no <see cref="CosmosClient"/> is registered.</summary>
 52    public string? ConnectionString { get; set; }
 53
 54    /// <summary>Cosmos database name. Required.</summary>
 55    public string? DatabaseName { get; set; }
 56
 57    /// <summary>Container storing one durable-flow ledger document per flow id.</summary>
 58    public string ContainerName { get; set; } = "asyncresponse_flow_state";
 59
 60    /// <summary>Partition-key path for the container. Default: <c>/flowId</c>.</summary>
 61    public string PartitionKeyPath { get; set; } = "/flowId";
 62
 63    /// <summary>Creates the database and container on first use.</summary>
 64    public bool AutoCreateContainer { get; set; } = true;
 65
 66    /// <summary>Optional throughput used when auto-creating the container.</summary>
 67    public int? Throughput { get; set; }
 68
 69    /// <summary>
 70    /// Maximum serialized flow-state size in bytes accepted by writes; oversized ledgers fail fast
 71    /// with an actionable error instead of the raw Cosmos 413 the executor would retry into the
 72    /// dead-letter queue. Default: 1.9 MB (headroom under Cosmos's 2 MB item cap for the sibling
 73    /// fields); <c>null</c> disables the guard.
 74    /// </summary>
 75    public long? MaxStateBytes { get; set; } = 1_900_000;
 76
 77    /// <summary>Validates option values and throws on misconfiguration.</summary>
 78    public void Validate()
 79    {
 80        if (string.IsNullOrWhiteSpace(DatabaseName))
 81            throw new InvalidOperationException($"{nameof(CosmosDurableFlowOptions)}.{nameof(DatabaseName)} must be conf
 82        if (string.IsNullOrWhiteSpace(ContainerName))
 83            throw new InvalidOperationException($"{nameof(CosmosDurableFlowOptions)}.{nameof(ContainerName)} must be con
 84        if (string.IsNullOrWhiteSpace(PartitionKeyPath) || !PartitionKeyPath.StartsWith("/", StringComparison.Ordinal))
 85            throw new InvalidOperationException($"{nameof(CosmosDurableFlowOptions)}.{nameof(PartitionKeyPath)} must sta
 86        if (Throughput is <= 0)
 87            throw new InvalidOperationException($"{nameof(CosmosDurableFlowOptions)}.{nameof(Throughput)} must be positi
 88        if (MaxStateBytes is <= 0)
 89            throw new InvalidOperationException($"{nameof(CosmosDurableFlowOptions)}.{nameof(MaxStateBytes)} must be pos
 90    }
 91}
 92
 93/// <summary>Azure Cosmos DB implementation of <see cref="IFlowStateStore"/>.</summary>
 94public sealed class CosmosFlowStateStore : IFlowStateStore, IDisposable
 95{
 96    // Time authority: this store keeps the app clock (DateTime.UtcNow) for expiry and lease
 97    // comparisons. Cosmos conditional writes (ETag preconditions) evaluate client-supplied
 98    // values only — there is no server-clock expression usable inside a point write — so the
 99    // read-check-replace cycles below compare against the app clock and rely on the ETag fence
 100    // for atomicity. Multi-node deployments should keep worker clocks synchronized well inside
 101    // the lease window. (The server's own TTL sweep, by contrast, runs on the service clock.)
 102    private readonly CosmosClient _client;
 103    private readonly CosmosDurableFlowOptions _options;
 104    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 105    private readonly bool _ownsClient;
 106    private bool _created;
 107
 108    public CosmosFlowStateStore(CosmosClient client, IOptions<CosmosDurableFlowOptions> options, bool ownsClient = false
 109    {
 110        _client = client;
 111        _options = options.Value;
 112        _options.Validate();
 113        _ownsClient = ownsClient;
 114    }
 115
 116    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 117    {
 118        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 119        var container = await GetContainerAsync(cancellationToken).ConfigureAwait(false);
 120
 121        try
 122        {
 123            var response = await container.ReadItemAsync<CosmosFlowStateDocument>(
 124                flowId,
 125                new PartitionKey(flowId),
 126                cancellationToken: cancellationToken).ConfigureAwait(false);
 127            var document = response.Resource;
 128            // Point reads have no predicate, so the expiry check happens client-side on the app
 129            // clock — see the time-authority note on this class.
 130            if (document.ExpiresAtUtc <= DateTime.UtcNow)
 131                return null;
 132
 133            return document.Revision is { } revision
 134                ? DurableFlowStoreShared.ReadState(flowId, document.StateJson, revision)
 135                : null;
 136        }
 137        catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.NotFound)
 138        {
 139            return null;
 140        }
 141    }
 142
 143    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 144    {
 145        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 146        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "Cosmos DB");
 147        var container = await GetContainerAsync(cancellationToken).ConfigureAwait(false);
 148
 149        // Bounded retry: documents carry native Cosmos TTL alongside the logical ExpiresAtUtc, so
 150        // the server's TTL sweep can physically purge an expired document between our create's
 151        // 409 and the follow-up read (or the conditional replace). A vanished conflicting
 152        // document means the slot is free — create again — not that a live competitor won.
 153        // The short growing delay lets one purge cycle finish instead of burning all attempts
 154        // inside the same inconsistency window (the Linux emulator is markedly worse here than
 155        // the service, but the race itself is real on both).
 156        for (var attempt = 0; attempt < 4; attempt++)
 157        {
 158            if (attempt > 0)
 159                await Task.Delay(TimeSpan.FromMilliseconds(50 * attempt), cancellationToken).ConfigureAwait(false);
 160
 161            var now = DateTime.UtcNow;
 162            var document = CreateDocument(flowId, stateJson, state.Revision, ttl, now);
 163            try
 164            {
 165                await container.CreateItemAsync(document, new PartitionKey(flowId), cancellationToken: cancellationToken
 166                return true;
 167            }
 168            catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.Conflict)
 169            {
 170                try
 171                {
 172                    var current = await container.ReadItemAsync<CosmosFlowStateDocument>(
 173                        flowId,
 174                        new PartitionKey(flowId),
 175                        cancellationToken: cancellationToken).ConfigureAwait(false);
 176                    if (current.Resource.ExpiresAtUtc > now)
 177                        return false;
 178
 179                    await container.ReplaceItemAsync(
 180                        document,
 181                        flowId,
 182                        new PartitionKey(flowId),
 183                        new ItemRequestOptions { IfMatchEtag = current.ETag },
 184                        cancellationToken).ConfigureAwait(false);
 185                    return true;
 186                }
 187                catch (CosmosException retryEx) when (retryEx.StatusCode == HttpStatusCode.PreconditionFailed)
 188                {
 189                    // The ETag moved under us. That is either a live writer winning the id or the
 190                    // TTL purge touching the expired document; only the next read can tell them
 191                    // apart, so retry instead of conceding — a live occupant surfaces as an
 192                    // unexpired read (false) on the next attempt, a purged one as a clean create.
 193                    continue;
 194                }
 195                catch (CosmosException retryEx) when (retryEx.StatusCode == HttpStatusCode.NotFound)
 196                {
 197                    // The expired document was TTL-purged after our 409 — the id is free again.
 198                    continue;
 199                }
 200            }
 201        }
 202
 203        return false;
 204    }
 205
 206    public async Task<bool> TryUpdateAsync(
 207        string flowId,
 208        FlowState state,
 209        long expectedRevision,
 210        TimeSpan ttl,
 211        string? leaseId = null,
 212        CancellationToken cancellationToken = default)
 213    {
 214        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 215        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "Cosmos DB");
 216        var container = await GetContainerAsync(cancellationToken).ConfigureAwait(false);
 217        for (var attempt = 0; attempt < 4; attempt++)
 218        {
 219            var now = DateTime.UtcNow;
 220            try
 221            {
 222                var current = await container.ReadItemAsync<CosmosFlowStateDocument>(
 223                    flowId,
 224                    new PartitionKey(flowId),
 225                    cancellationToken: cancellationToken).ConfigureAwait(false);
 226                var document = current.Resource;
 227                if (document.ExpiresAtUtc <= now || document.Revision != expectedRevision)
 228                    return false;
 229                if (leaseId is not null && (document.LeaseId != leaseId || document.LeaseExpiresAtUtc <= now))
 230                    return false;
 231
 232                document.StateJson = stateJson;
 233                document.ExpiresAtUtc = DurableFlowStoreShared.AddSaturating(now, ttl);
 234                document.UpdatedAtUtc = now;
 235                document.Revision = state.Revision;
 236                document.Ttl = CosmosTtlSeconds(ttl);
 237                await container.ReplaceItemAsync(
 238                    document,
 239                    flowId,
 240                    new PartitionKey(flowId),
 241                    new ItemRequestOptions { IfMatchEtag = current.ETag },
 242                    cancellationToken).ConfigureAwait(false);
 243                return true;
 244            }
 245            catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.NotFound)
 246            {
 247                return false;
 248            }
 249            catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.PreconditionFailed)
 250            {
 251                // Lease renewal also replaces the document and changes its ETag without changing
 252                // the ledger revision. Re-read and retry so that benign race is not reported as a
 253                // lost execution lease; a real state race fails the revision check above.
 254            }
 255        }
 256
 257        return false;
 258    }
 259
 260    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 261        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 262
 263    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 264        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 265
 266    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 267    {
 268        var container = await GetContainerAsync(cancellationToken).ConfigureAwait(false);
 269        for (var attempt = 0; attempt < 4; attempt++)
 270        {
 271            try
 272            {
 273                var current = await container.ReadItemAsync<CosmosFlowStateDocument>(
 274                    flowId,
 275                    new PartitionKey(flowId),
 276                    cancellationToken: cancellationToken).ConfigureAwait(false);
 277                if (current.Resource.LeaseId != leaseId)
 278                    return;
 279
 280                current.Resource.LeaseId = null;
 281                current.Resource.LeaseExpiresAtUtc = null;
 282                // Every replace refreshes _ts — the anchor the server-side TTL counts from — so
 283                // re-persisting the stored full-window ttl would restart the physical-retention
 284                // countdown and decouple it from the logical ExpiresAtUtc. Rewrite it from the
 285                // remaining logical window instead. (Checkpoints recompute both together in
 286                // TryUpdateAsync; only the lease paths replace without moving ExpiresAtUtc.)
 287                current.Resource.Ttl = CosmosTtlSeconds(current.Resource.ExpiresAtUtc, DateTime.UtcNow);
 288                await container.ReplaceItemAsync(
 289                    current.Resource,
 290                    flowId,
 291                    new PartitionKey(flowId),
 292                    new ItemRequestOptions { IfMatchEtag = current.ETag },
 293                    cancellationToken).ConfigureAwait(false);
 294                return;
 295            }
 296            catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.NotFound)
 297            {
 298                return;
 299            }
 300            catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.PreconditionFailed)
 301            {
 302            }
 303        }
 304    }
 305
 306    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 307    {
 308        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 309        var container = await GetContainerAsync(cancellationToken).ConfigureAwait(false);
 310
 311        try
 312        {
 313            await container.DeleteItemAsync<CosmosFlowStateDocument>(
 314                flowId,
 315                new PartitionKey(flowId),
 316                cancellationToken: cancellationToken).ConfigureAwait(false);
 317            return true;
 318        }
 319        catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.NotFound)
 320        {
 321            return false;
 322        }
 323    }
 324
 325    private async Task<Container> GetContainerAsync(CancellationToken cancellationToken)
 326    {
 327        if (!_created)
 328            await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 329
 330        return _client.GetContainer(_options.DatabaseName, _options.ContainerName);
 331    }
 332
 333    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 334    {
 335        if (_created)
 336            return;
 337
 338        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 339        try
 340        {
 341            if (_created)
 342                return;
 343
 344            ContainerResponse container;
 345            if (_options.AutoCreateContainer)
 346            {
 347                var database = await _client.CreateDatabaseIfNotExistsAsync(
 348                    _options.DatabaseName,
 349                    cancellationToken: cancellationToken).ConfigureAwait(false);
 350
 351                // DefaultTimeToLive = -1 enables per-item TTL without a container-wide default.
 352                var properties = new ContainerProperties(_options.ContainerName, _options.PartitionKeyPath)
 353                {
 354                    DefaultTimeToLive = -1
 355                };
 356                container = await database.Database.CreateContainerIfNotExistsAsync(
 357                    properties,
 358                    _options.Throughput,
 359                    cancellationToken: cancellationToken).ConfigureAwait(false);
 360            }
 361            else
 362            {
 363                container = await _client
 364                    .GetContainer(_options.DatabaseName, _options.ContainerName)
 365                    .ReadContainerAsync(cancellationToken: cancellationToken)
 366                    .ConfigureAwait(false);
 367            }
 368
 369            if (!string.Equals(container.Resource.PartitionKeyPath, _options.PartitionKeyPath, StringComparison.Ordinal)
 370                throw new InvalidOperationException(
 371                    $"Cosmos container '{_options.ContainerName}' uses partition key '{container.Resource.PartitionKeyPa
 372                    $"but '{_options.PartitionKeyPath}' is required.");
 373            if (container.Resource.DefaultTimeToLive is null)
 374                throw new InvalidOperationException(
 375                    $"Cosmos container '{_options.ContainerName}' does not have TTL enabled. " +
 376                    "Enable container TTL (DefaultTimeToLive = -1) before using it for durable flows.");
 377
 378            ValidateHostSerializer();
 379            _created = true;
 380        }
 381        finally
 382        {
 383            _ensureGate.Release();
 384        }
 385    }
 386
 387    /// <summary>
 388    /// Fails provisioning fast when a host-registered <see cref="CosmosClient"/> carries a custom
 389    /// serializer that does not honor the flow-state document's JSON property names.
 390    /// <see cref="CosmosFlowStateDocument"/> is attributed for both Newtonsoft.Json
 391    /// (<c>[JsonProperty]</c>) and System.Text.Json (<c>[JsonPropertyName]</c>), so the SDK's
 392    /// default serializer and STJ-based serializers both map correctly; a serializer honoring
 393    /// neither would write documents whose <c>id</c> Cosmos rejects — or, worse, whose fields
 394    /// silently round-trip as nulls. The probe serializes a sentinel document through the host's
 395    /// serializer and verifies the wire property names survive.
 396    /// </summary>
 397    private void ValidateHostSerializer()
 398    {
 399        if (_client.ClientOptions?.Serializer is not { } serializer)
 400            return; // The SDK default serializer honors [JsonProperty]; nothing to probe.
 401
 402        var now = DateTime.UtcNow;
 403        using var stream = serializer.ToStream(new CosmosFlowStateDocument
 404        {
 405            Id = "asyncresponse-serializer-probe",
 406            FlowId = "asyncresponse-serializer-probe",
 407            StateJson = "{}",
 408            ExpiresAtUtc = now,
 409            UpdatedAtUtc = now,
 410            Revision = 0,
 411            LeaseId = "probe",
 412            LeaseExpiresAtUtc = now,
 413            Ttl = 1
 414        });
 415        using var probe = System.Text.Json.JsonDocument.Parse(stream);
 416        if (!probe.RootElement.TryGetProperty("id", out _) || !probe.RootElement.TryGetProperty("leaseExpiresAtUtc", out
 417        {
 418            throw new InvalidOperationException(
 419                $"The registered CosmosClient's serializer ({serializer.GetType().Name}) does not honor the durable-flow
 420                "JSON property names ('id', 'flowId', 'leaseExpiresAtUtc', ...). Flow-state documents would be written w
 421                "property names and could not be read back. Configure the serializer to honor Newtonsoft.Json [JsonPrope
 422                "System.Text.Json [JsonPropertyName] attributes, or let the SDK use its default serializer.");
 423        }
 424    }
 425
 426    private async Task<bool> UpdateLeaseAsync(
 427        string flowId,
 428        string leaseId,
 429        TimeSpan leaseDuration,
 430        bool acquire,
 431        CancellationToken cancellationToken)
 432    {
 433        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 434        ArgumentException.ThrowIfNullOrWhiteSpace(leaseId);
 435        if (leaseDuration <= TimeSpan.Zero)
 436            throw new ArgumentOutOfRangeException(nameof(leaseDuration));
 437
 438        var container = await GetContainerAsync(cancellationToken).ConfigureAwait(false);
 439        for (var attempt = 0; attempt < 4; attempt++)
 440        {
 441            var now = DateTime.UtcNow;
 442            try
 443            {
 444                var current = await container.ReadItemAsync<CosmosFlowStateDocument>(
 445                    flowId,
 446                    new PartitionKey(flowId),
 447                    cancellationToken: cancellationToken).ConfigureAwait(false);
 448                var document = current.Resource;
 449                if (document.ExpiresAtUtc <= now || document.Revision is null)
 450                    return false;
 451                if (acquire)
 452                {
 453                    if (document.LeaseId is not null && document.LeaseId != leaseId && document.LeaseExpiresAtUtc > now)
 454                        return false;
 455                }
 456                else if (document.LeaseId != leaseId || document.LeaseExpiresAtUtc <= now)
 457                {
 458                    return false;
 459                }
 460
 461                document.LeaseId = leaseId;
 462                document.LeaseExpiresAtUtc = DurableFlowStoreShared.AddSaturating(now, leaseDuration);
 463                // Same _ts realignment as ReleaseLeaseAsync: a lease heartbeat replaces the
 464                // document without moving ExpiresAtUtc, so it must not restart the server TTL's
 465                // full retention window.
 466                document.Ttl = CosmosTtlSeconds(document.ExpiresAtUtc, now);
 467                await container.ReplaceItemAsync(
 468                    document,
 469                    flowId,
 470                    new PartitionKey(flowId),
 471                    new ItemRequestOptions { IfMatchEtag = current.ETag },
 472                    cancellationToken).ConfigureAwait(false);
 473                return true;
 474            }
 475            catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.NotFound)
 476            {
 477                return false;
 478            }
 479            catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.PreconditionFailed)
 480            {
 481            }
 482        }
 483
 484        return false;
 485    }
 486
 487    private static CosmosFlowStateDocument CreateDocument(string flowId, string stateJson, long revision, TimeSpan ttl, 
 488        => new()
 489        {
 490            Id = flowId,
 491            FlowId = flowId,
 492            StateJson = stateJson,
 493            ExpiresAtUtc = DurableFlowStoreShared.AddSaturating(now, ttl),
 494            UpdatedAtUtc = now,
 495            Revision = revision,
 496            // Cosmos reaps the item itself once container TTL is enabled. Ceiling keeps the
 497            // server-side TTL from being shorter than the requested duration.
 498            Ttl = CosmosTtlSeconds(ttl)
 499        };
 500
 501    /// <summary>Per-item TTL in whole seconds, rounded up and saturated at int.MaxValue (~68 years) for absurd expiries
 502    private static int CosmosTtlSeconds(TimeSpan ttl)
 503        => (int)Math.Min(Math.Ceiling(ttl.TotalSeconds), int.MaxValue);
 504
 505    /// <summary>
 506    /// Remaining per-item TTL in whole seconds until <paramref name="expiresAtUtc"/>, rounded up
 507    /// and floored at 1 (Cosmos rejects 0). Used by replaces that keep the logical expiry in place:
 508    /// an already-due document collapses to the shortest legal TTL so the next sweep purges it
 509    /// instead of the replace granting it a fresh retention window.
 510    /// </summary>
 511    private static int CosmosTtlSeconds(DateTime expiresAtUtc, DateTime now)
 512        => (int)Math.Min(Math.Max(Math.Ceiling((expiresAtUtc - now).TotalSeconds), 1), int.MaxValue);
 513
 514    /// <summary>Disposes the Cosmos client when the store created (and therefore owns) it.</summary>
 515    public void Dispose()
 516    {
 517        _ensureGate.Dispose();
 518        if (_ownsClient)
 519            _client.Dispose();
 520    }
 521}
 522
 523/// <summary>
 524/// One durable-flow ledger document. Attributed for BOTH Newtonsoft.Json and System.Text.Json:
 525/// the Cosmos SDK's default serializer is Newtonsoft-based, but hosts may register a
 526/// <see cref="CosmosClient"/> with an STJ-based serializer — with single-stack attributes such a
 527/// client would silently write PascalCase property names (breaking <c>id</c> and every read
 528/// back). <see cref="CosmosFlowStateStore"/> additionally probes custom serializers at
 529/// provisioning time and fails fast when neither attribute set is honored.
 530/// </summary>
 531internal sealed class CosmosFlowStateDocument
 532{
 533    [JsonProperty("id")]
 534    [System.Text.Json.Serialization.JsonPropertyName("id")]
 535    public string Id { get; set; } = "";
 536
 537    [JsonProperty("flowId")]
 538    [System.Text.Json.Serialization.JsonPropertyName("flowId")]
 539    public string FlowId { get; set; } = "";
 540
 541    [JsonProperty("stateJson")]
 542    [System.Text.Json.Serialization.JsonPropertyName("stateJson")]
 543    public string StateJson { get; set; } = "";
 544
 545    [JsonProperty("expiresAtUtc")]
 546    [System.Text.Json.Serialization.JsonPropertyName("expiresAtUtc")]
 547    public DateTime ExpiresAtUtc { get; set; }
 548
 549    [JsonProperty("updatedAtUtc")]
 550    [System.Text.Json.Serialization.JsonPropertyName("updatedAtUtc")]
 551    public DateTime UpdatedAtUtc { get; set; }
 552
 553    [JsonProperty("revision")]
 554    [System.Text.Json.Serialization.JsonPropertyName("revision")]
 555    public long? Revision { get; set; }
 556
 557    [JsonProperty("leaseId", NullValueHandling = NullValueHandling.Ignore)]
 558    [System.Text.Json.Serialization.JsonPropertyName("leaseId")]
 559    [System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritin
 560    public string? LeaseId { get; set; }
 561
 562    [JsonProperty("leaseExpiresAtUtc", NullValueHandling = NullValueHandling.Ignore)]
 563    [System.Text.Json.Serialization.JsonPropertyName("leaseExpiresAtUtc")]
 564    [System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritin
 565    public DateTime? LeaseExpiresAtUtc { get; set; }
 566
 567    /// <summary>Cosmos per-item TTL in seconds; honored once the container enables TTL.</summary>
 568    [JsonProperty("ttl", NullValueHandling = NullValueHandling.Ignore)]
 569    [System.Text.Json.Serialization.JsonPropertyName("ttl")]
 570    [System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritin
 571    public int? Ttl { get; set; }
 572}
 573}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

WithCosmosDurableFlows(Microsoft.Extensions.DependencyInjection.AsyncResponseRegistrationBuilder, System.Action<AsyncResponse.DurableFlows.Cosmos.CosmosDurableFlowOptions>)