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

Information
Class: AsyncResponse.DurableFlows.Internal.DurableFlowStoreShared
Assembly: AsyncResponse.DurableFlows.Oracle
File(s): /_/src/DurableFlows/Shared/DurableFlowStoreShared.cs
Line coverage
100%
Covered lines: 126
Uncovered lines: 0
Coverable lines: 126
Total lines: 396
Line coverage: 100%
Branch coverage
97%
Covered branches: 76
Total branches: 78
Branch coverage: 97.4%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.cctor()100%11100%
PruneQuietlyAsync()87.5%88100%
ValidatePruneBudget(...)100%22100%
ValidateCreate(...)100%22100%
ValidateUpdate(...)100%44100%
ValidateLeaseArgs(...)100%22100%
LeaseObservation(...)100%66100%
Serialize(...)100%11100%
SerializeBounded(...)100%44100%
ReadState(...)100%44100%
AddSaturating(...)100%22100%
AddSaturating(...)100%22100%
ServerClockTtl(...)100%22100%
ServerClockTtlMilliseconds(...)100%11100%
Deserialize(...)100%11100%
ShouldPrune(...)100%44100%
SchemaLockKey(...)100%22100%
SchemaLockResource(...)100%11100%
ValidateConnectionString(...)100%22100%
ValidateMaxStateBytes(...)100%44100%
OpenConnectionAsync()100%1171.42%
ValidateIdentifier(...)100%1616100%
DerivedName(...)83.33%66100%
ValidateWrite(...)100%66100%

File(s)

/_/src/DurableFlows/Shared/DurableFlowStoreShared.cs

#LineLine coverage
 1using System.Data.Common;
 2using System.Diagnostics;
 3using System.Text;
 4using Microsoft.Extensions.Logging;
 5
 6namespace AsyncResponse.DurableFlows.Internal;
 7
 8internal static class DurableFlowStoreShared
 9{
 10    /// <summary>
 11    /// Rows one prune statement deletes. Every relational store deletes in batches of this size:
 12    /// an unbatched DELETE over a large expired backlog holds row locks and bloats one transaction
 13    /// for the unlucky create that triggered the prune.
 14    /// </summary>
 15    public const int PruneBatchSize = 1000;
 16
 17    /// <summary>
 18    /// The default <c>PruneBudget</c>: wall-clock time one opportunistic prune may spend draining
 19    /// batches after the first. Two seconds at ~1000 rows per batch drains tens of thousands of
 20    /// expired rows per interval on an ordinary database, against the ~3 rows/second a single
 21    /// batch per five-minute interval sustained — which any instance creating more than that fell
 22    /// behind forever.
 23    /// </summary>
 624    public static readonly TimeSpan DefaultPruneBudget = TimeSpan.FromSeconds(2);
 25
 26    /// <summary>
 27    /// Runs an opportunistic prune so that its failure never fails the primitive it rides on,
 28    /// draining <see cref="PruneBatchSize"/>-row batches until a batch comes back short (the
 29    /// backlog is gone) or <paramref name="budget"/> lapses. The first batch always runs, so a
 30    /// zero budget is the historical single-batch policy. Awaited bare inside
 31    /// <c>TryCreateAsync</c>, a prune chosen as the deadlock victim (1205) or hitting a lock-wait
 32    /// timeout against the store's own live checkpoint traffic failed <c>StartAsync</c> for a flow
 33    /// whose row would have been created without incident — and <see cref="ShouldPrune"/> had
 34    /// already consumed the interval, so it was not retried either. Loads filter on expiry, so a
 35    /// skipped prune costs nothing but disk until the next interval. The outcome is never silent:
 36    /// deleted rows, a lapsed budget with rows remaining, and failures are counted on the
 37    /// <c>AsyncResponse</c> meter and logged when the store has a logger. Cancellation still
 38    /// propagates.
 39    /// </summary>
 40    /// <param name="pruneBatch">Deletes one batch and returns the rows it deleted.</param>
 41    /// <param name="budget">Wall-clock budget for batches after the first.</param>
 42    /// <param name="providerName">Metric/log tag for the store ("PostgreSQL", "SQL Server", …).</param>
 43    /// <param name="logger">The store's logger when DI supplied one.</param>
 44    public static async Task PruneQuietlyAsync(Func<Task<int>> pruneBatch, TimeSpan budget, string providerName, ILogger
 45    {
 15746        var started = Stopwatch.GetTimestamp();
 15747        var deleted = 0L;
 15748        var batches = 0;
 49        try
 50        {
 51            while (true)
 52            {
 16153                var batchDeleted = await pruneBatch().ConfigureAwait(false);
 14654                batches++;
 14655                deleted += Math.Max(batchDeleted, 0);
 14656                if (batchDeleted < PruneBatchSize)
 57                    break;
 58
 659                if (Stopwatch.GetElapsedTime(started) >= budget)
 60                {
 261                    AsyncResponseDiagnostics.RecordFlowStatePruneBudgetExhausted(providerName);
 262                    logger?.LogWarning(
 263                        "{Provider} durable-flow prune deleted {Deleted} expired rows in {Batches} batches and stopped a
 264                        providerName, deleted, batches, budget);
 65                    break;
 66                }
 67            }
 68
 14269            AsyncResponseDiagnostics.RecordFlowStatePruned(providerName, deleted);
 14270        }
 871        catch (OperationCanceledException)
 72        {
 873            AsyncResponseDiagnostics.RecordFlowStatePruned(providerName, deleted);
 874            throw;
 75        }
 776        catch (Exception ex)
 77        {
 78            // Opportunistic maintenance; the next interval retries — but never silently.
 779            AsyncResponseDiagnostics.RecordFlowStatePruned(providerName, deleted);
 780            AsyncResponseDiagnostics.RecordFlowStatePruneFailure(providerName);
 781            logger?.LogWarning(
 782                ex,
 783                "{Provider} durable-flow prune failed after deleting {Deleted} expired rows in {Batches} batches; the fl
 784                providerName, deleted, batches);
 785        }
 14986    }
 87
 88    /// <summary>A <c>PruneBudget</c> is a non-negative duration; zero means a single batch per interval.</summary>
 89    public static void ValidatePruneBudget(TimeSpan budget, string optionsName)
 90    {
 23291        if (budget < TimeSpan.Zero)
 292            throw new InvalidOperationException($"{optionsName}.PruneBudget cannot be negative (zero limits each prune t
 23093    }
 94
 95    /// <summary>
 96    /// Upper bound for TTL values handed to server-clock date arithmetic (~68 years). SQL Server's
 97    /// <c>DATEADD</c> takes <c>int</c> seconds, and MySQL/Oracle datetime types stop at year 9999,
 98    /// so an absurd <see cref="DurableFlowOptions.StateExpiry"/> (for example
 99    /// <see cref="TimeSpan.MaxValue"/>) would overflow inside the database. Clamping mirrors
 100    /// <see cref="AddSaturating(DateTime, TimeSpan)"/> on the client side: huge expiries saturate
 101    /// to "effectively never" instead of failing every write.
 102    /// </summary>
 6103    private static readonly TimeSpan MaxServerClockTtl = TimeSpan.FromSeconds(int.MaxValue);
 104
 105    public static void ValidateCreate(string flowId, FlowState state, TimeSpan ttl)
 106    {
 455107        ValidateWrite(flowId, state, ttl);
 444108        if (state.Revision != 0)
 2109            throw new ArgumentException("A new flow ledger must start at revision zero.", nameof(state));
 442110    }
 111
 112    public static void ValidateUpdate(string flowId, FlowState state, long expectedRevision, TimeSpan ttl)
 113    {
 883114        ValidateWrite(flowId, state, ttl);
 883115        if (expectedRevision < 0)
 2116            throw new ArgumentOutOfRangeException(nameof(expectedRevision), "The expected revision cannot be negative.")
 881117        if (state.Revision != checked(expectedRevision + 1))
 2118            throw new ArgumentException("The new flow-state revision must increment the expected revision by one.", name
 877119    }
 120
 121    /// <summary>
 122    /// Argument preamble shared by every store's lease acquire/renew path. The stores pass their
 123    /// own parameters straight through, so the thrown <c>ParamName</c>s ("flowId", "leaseId",
 124    /// "leaseDuration") match the public <c>IFlowStateStore</c> signatures exactly.
 125    /// </summary>
 126    public static void ValidateLeaseArgs(string flowId, string leaseId, TimeSpan leaseDuration)
 127    {
 175128        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 173129        ArgumentException.ThrowIfNullOrWhiteSpace(leaseId);
 171130        if (leaseDuration <= TimeSpan.Zero)
 6131            throw new ArgumentOutOfRangeException(nameof(leaseDuration));
 165132    }
 133
 134    /// <summary>
 135    /// Shapes one persisted lease pair into the <see cref="IFlowStateStore.ObserveLeaseAsync"/>
 136    /// answer, identically in all nine stores. The pair is reported RAW: nothing here (or in the
 137    /// reads that feed it) compares the expiry with a clock, because the engine's liveness proof
 138    /// is that two observations differ, and an expired lease that nobody took over must keep
 139    /// reading as the same lease until someone acquires or releases it.
 140    /// <para>
 141    /// A <c>null</c> <paramref name="leaseId"/> — never leased, released, or no row at all — is
 142    /// <see cref="FlowLeaseObservation.Unheld"/>, never <c>null</c>: <c>null</c> is reserved for
 143    /// "this store cannot report leases".
 144    /// </para>
 145    /// <para>
 146    /// Every store persists the lease expiry as a UTC instant, but the drivers disagree about the
 147    /// <see cref="DateTimeKind"/> they hand back (zone-less SQL columns read as
 148    /// <see cref="DateTimeKind.Unspecified"/>; a legacy-timestamp Npgsql or a custom Cosmos
 149    /// serializer can produce <see cref="DateTimeKind.Local"/>). The observation always carries
 150    /// <see cref="DateTimeKind.Utc"/> with the ticks the store kept, so two renewals compare
 151    /// strictly increasing whichever driver read them.
 152    /// </para>
 153    /// </summary>
 154    public static FlowLeaseObservation LeaseObservation(string? leaseId, DateTime? leaseExpiresAt)
 155    {
 22156        if (leaseId is null)
 6157            return FlowLeaseObservation.Unheld;
 158
 16159        return new FlowLeaseObservation(
 16160            leaseId,
 16161            leaseExpiresAt is { } expiry
 16162                ? expiry.Kind == DateTimeKind.Local ? expiry.ToUniversalTime() : DateTime.SpecifyKind(expiry, DateTimeKi
 16163                : null);
 164    }
 165
 166    /// <summary>
 167    /// The ledger has exactly ONE wire format: Core's <c>FlowStateJson</c> (source-generated
 168    /// metadata, nulls omitted on write, resolved through the <c>AsyncResponseJson</c> chain so
 169    /// <c>AsyncResponseJsonSerialization.RegisterResolver</c> — the documented trim/AOT seam —
 170    /// reaches every store). Serialize and Deserialize both delegate there, so a ledger written
 171    /// through any provider store always loads through any other, byte for byte.
 172    /// </summary>
 1197173    public static string Serialize(FlowState state) => FlowStateJson.Serialize(state);
 174
 175    /// <summary>
 176    /// Serializes a ledger for a full-state write and enforces the store's <c>MaxStateBytes</c>
 177    /// budget. Without the guard an oversized ledger surfaces as the provider's opaque payload
 178    /// error (DynamoDB 400 KB item cap, Cosmos 2 MB, MongoDB 16 MB) which the executor retries
 179    /// into the dead-letter queue with no hint at the real cause. Throwing here keeps the same
 180    /// at-least-once semantics (run fails → retries → DLQ = operator alarm) but names the cause.
 181    /// </summary>
 182    /// <exception cref="FlowStateTooLargeException">The serialized state exceeds <paramref name="maxStateBytes"/>.</exc
 183    public static string SerializeBounded(string flowId, FlowState state, long? maxStateBytes, string providerName)
 184    {
 1189185        var json = Serialize(state);
 1189186        if (maxStateBytes is { } limit)
 187        {
 8188            long size = Encoding.UTF8.GetByteCount(json);
 8189            if (size > limit)
 4190                throw new FlowStateTooLargeException(flowId, size, limit, providerName);
 191        }
 192
 1185193        return json;
 194    }
 195
 196    /// <summary>
 197    /// Materializes a loaded ledger row.
 198    /// <para>
 199    /// The row is never executed as anything but what it consistently says it is: a revision
 200    /// inside the JSON that disagrees with the row's own revision column, or a ledger whose
 201    /// <c>FlowId</c> is not the key it was loaded under (a row copied or restored under the
 202    /// wrong key), is refused — the read-side mirror of the write-side key/identity validation
 203    /// in <see cref="ValidateCreate"/>.
 204    /// </para>
 205    /// <para>
 206    /// Refused means <see cref="FlowStateUnreadableException"/>, never <c>null</c>. Every built-in
 207    /// store reads the JSON and the revision from ONE row or document, so a disagreement inside
 208    /// that snapshot is an inconsistent — corrupt, hand-edited, mis-restored — ledger that is
 209    /// physically present, not proof the run is gone. A <c>null</c> here told the executor to
 210    /// acknowledge the wake-up as belonging to a deleted flow, and the run behind the row lost
 211    /// its only wake-up while its row sat in the table. Unreadable JSON and an unknown schema
 212    /// version throw for the same reason (see the exception's remarks); the delivery rides the
 213    /// transport's retry and dead-letter path, which is the operator alarm.
 214    /// </para>
 215    /// </summary>
 216    /// <exception cref="FlowStateUnreadableException">The row is present but uninterpretable or inconsistent.</exceptio
 217    public static FlowState? ReadState(string flowId, string stateJson, long revision)
 218    {
 681219        var state = Deserialize(stateJson, flowId);
 679220        if (state.Revision != revision)
 221        {
 3222            throw new FlowStateUnreadableException(
 3223                flowId,
 3224                $"its stored revision is {revision} but the revision inside its JSON is {state.Revision}");
 225        }
 226
 676227        if (!string.Equals(state.FlowId, flowId, StringComparison.Ordinal))
 2228            throw new FlowStateUnreadableException(flowId, "the flow id inside its JSON is not the id it is stored under
 229
 674230        return state;
 231    }
 232
 233    /// <summary>
 234    /// <paramref name="instant"/> + <paramref name="ttl"/>, saturating at
 235    /// <see cref="DateTime.MaxValue"/> instead of throwing: an absurd
 236    /// <see cref="DurableFlowOptions.StateExpiry"/> then means "effectively never expires" rather
 237    /// than failing every write with an <see cref="ArgumentOutOfRangeException"/>.
 238    /// </summary>
 239    public static DateTime AddSaturating(DateTime instant, TimeSpan ttl)
 4240        => ttl > DateTime.MaxValue - instant ? DateTime.MaxValue : instant + ttl;
 241
 242    /// <inheritdoc cref="AddSaturating(DateTime, TimeSpan)"/>
 243    public static DateTimeOffset AddSaturating(DateTimeOffset instant, TimeSpan ttl)
 4244        => ttl > DateTimeOffset.MaxValue - instant ? DateTimeOffset.MaxValue : instant + ttl;
 245
 246    /// <summary>TTL clamped for server-clock date arithmetic; see <see cref="MaxServerClockTtl"/>.</summary>
 247    public static TimeSpan ServerClockTtl(TimeSpan ttl)
 1386248        => ttl > MaxServerClockTtl ? MaxServerClockTtl : ttl;
 249
 250    /// <summary>Whole milliseconds of <see cref="ServerClockTtl"/>, for stores that bind the TTL as a number.</summary>
 251    public static long ServerClockTtlMilliseconds(TimeSpan ttl)
 1382252        => (long)ServerClockTtl(ttl).TotalMilliseconds;
 253
 254    /// <inheritdoc cref="FlowStateJson.Deserialize"/>
 697255    public static FlowState Deserialize(string json, string flowId) => FlowStateJson.Deserialize(json, flowId);
 256
 257    /// <summary>
 258    /// Throttles opportunistic expired-state pruning: returns <c>true</c> at most once per
 259    /// <paramref name="interval"/> (a non-positive interval prunes on every operation, matching the
 260    /// channel packages). Loads already filter on expiry, so throttling never affects correctness.
 261    /// </summary>
 262    public static bool ShouldPrune(ref long lastTicks, TimeSpan interval)
 263    {
 304264        if (interval <= TimeSpan.Zero)
 2265            return true;
 266
 302267        var now = DateTime.UtcNow.Ticks;
 302268        var last = Interlocked.Read(ref lastTicks);
 302269        return now - last >= interval.Ticks
 302270            && Interlocked.CompareExchange(ref lastTicks, now, last) == last;
 271    }
 272
 273    /// <summary>
 274    /// Advisory-lock key for schema DDL, derived exactly like the channel/transport packages
 275    /// (FNV-1a over <c>asyncresponse:ddl:{schemaName}</c>) so flow-store DDL serializes with any
 276    /// channel/transport DDL running against the same schema.
 277    /// </summary>
 278    public static long SchemaLockKey(string schemaName)
 279    {
 280        const ulong offset = 14695981039346656037UL;
 281        const ulong prime = 1099511628211UL;
 6282        var hash = offset;
 380283        foreach (var b in Encoding.UTF8.GetBytes(SchemaLockResource(schemaName)))
 284        {
 184285            hash ^= b;
 184286            hash *= prime;
 287        }
 288
 6289        return unchecked((long)hash);
 290    }
 291
 292    /// <summary>SQL Server <c>sp_getapplock</c> resource name for schema DDL (shared with the channel/transport package
 293    public static string SchemaLockResource(string schemaName)
 8294        => $"asyncresponse:ddl:{schemaName}";
 295
 296    /// <summary>Connection-string guard shared by the stores that own their connections.</summary>
 297    public static void ValidateConnectionString(string? connectionString, string optionsTypeName)
 298    {
 240299        if (string.IsNullOrWhiteSpace(connectionString))
 2300            throw new InvalidOperationException($"{optionsTypeName}.ConnectionString must be configured.");
 238301    }
 302
 303    /// <summary><c>MaxStateBytes</c> guard shared by all nine stores (null disables the budget).</summary>
 304    public static void ValidateMaxStateBytes(long? maxStateBytes, string optionsTypeName)
 305    {
 234306        if (maxStateBytes is <= 0)
 2307            throw new InvalidOperationException($"{optionsTypeName}.MaxStateBytes must be positive when configured.");
 232308    }
 309
 310    /// <summary>
 311    /// Opens a fresh provider connection, disposing it when open fails — an ADO.NET connection
 312    /// that failed to open still holds its allocation until disposed, and the caller never
 313    /// receives it.
 314    /// </summary>
 315    public static async Task<TConnection> OpenConnectionAsync<TConnection>(
 316        string? connectionString,
 317        CancellationToken cancellationToken)
 318        where TConnection : DbConnection, new()
 319    {
 2470320        var connection = new TConnection { ConnectionString = connectionString };
 321        try
 322        {
 2470323            await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
 2466324            return connection;
 325        }
 4326        catch
 327        {
 4328            await connection.DisposeAsync().ConfigureAwait(false);
 4329            throw;
 330        }
 2466331    }
 332
 333    public static void ValidateIdentifier(string? value, string optionName, string providerName, int identifierCap = 0)
 334    {
 252335        if (string.IsNullOrWhiteSpace(value))
 2336            throw new InvalidOperationException($"{optionName} must be configured.");
 337
 250338        if (!(char.IsAsciiLetter(value[0]) || value[0] == '_'))
 2339            throw new InvalidOperationException($"{optionName} '{value}' must be a simple {providerName} identifier (let
 340
 10774341        foreach (var c in value)
 342        {
 5140343            if (!(char.IsAsciiLetterOrDigit(c) || c == '_'))
 2344                throw new InvalidOperationException($"{optionName} '{value}' must be a simple {providerName} identifier 
 345        }
 346
 246347        if (identifierCap > 0 && value.Length > identifierCap)
 2348            throw new InvalidOperationException(
 2349                $"{optionName} '{value}' is {value.Length} characters; {providerName} identifiers are limited to {identi
 244350    }
 351
 352    /// <summary>
 353    /// Derived object name with suffix space RESERVED before the provider's identifier cap:
 354    /// truncating the whole "{table}{suffix}" lets a maximum-length table name derive its own
 355    /// name — on providers where indexes share the table namespace the DDL is then silently
 356    /// skipped, on the rest it fails outright.
 357    /// <para>
 358    /// Kept in step with <c>AsyncResponse.Internal.RelationalNamePlan.DerivedName</c>, which is
 359    /// the same rule for the channel and transport packages. The two cannot be one method: this
 360    /// file is source-linked into all nine flow stores, and RelationalNamePlan is linked only
 361    /// into the four relational channel/transport packages that need a name plan.
 362    /// </para>
 363    /// </summary>
 364    /// <exception cref="ArgumentOutOfRangeException">
 365    /// <paramref name="suffix"/> leaves no room for a stem inside <paramref name="identifierCap"/>.
 366    /// Guarded explicitly: the slice below would otherwise take a negative length and fail schema
 367    /// creation with a bare index-out-of-range naming neither the suffix nor the cap.
 368    /// </exception>
 369    public static string DerivedName(string tableName, string suffix, int identifierCap)
 370    {
 383371        if (identifierCap <= 0 || tableName.Length + suffix.Length <= identifierCap)
 373372            return tableName + suffix;
 373
 10374        if (suffix.Length >= identifierCap)
 375        {
 2376            throw new ArgumentOutOfRangeException(
 2377                nameof(suffix),
 2378                $"The derived-name suffix '{suffix}' is {suffix.Length} characters, which leaves no room for a table ste
 2379                $"the {identifierCap}-character identifier limit. Shorten the suffix.");
 380        }
 381
 8382        return tableName[..(identifierCap - suffix.Length)] + suffix;
 383    }
 384
 385    private static void ValidateWrite(string flowId, FlowState state, TimeSpan ttl)
 386    {
 1338387        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 1336388        ArgumentNullException.ThrowIfNull(state);
 1334389        if (!string.Equals(state.FlowId, flowId, StringComparison.Ordinal))
 2390            throw new ArgumentException("The flow state id must match the store key.", nameof(state));
 1332391        if (state.SchemaVersion != FlowStateSchema.Current)
 3392            throw new ArgumentException("The flow state must use the current schema version.", nameof(state));
 1329393        if (ttl <= TimeSpan.Zero)
 2394            throw new ArgumentOutOfRangeException(nameof(ttl), "TTL must be greater than zero.");
 1327395    }
 396}