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

Information
Class: AsyncResponse.DurableFlows.Oracle.OracleDurableFlowOptions
Assembly: AsyncResponse.DurableFlows.Oracle
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.Oracle/OracleDurableFlows.cs
Line coverage
100%
Covered lines: 9
Uncovered lines: 0
Coverable lines: 9
Total lines: 352
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
.ctor()100%11100%
Validate()100%66100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.Oracle/OracleDurableFlows.cs

#LineLine coverage
 1using AsyncResponse;
 2using AsyncResponse.DurableFlows.Internal;
 3using AsyncResponse.DurableFlows.Oracle;
 4using Microsoft.Extensions.DependencyInjection.Extensions;
 5using Microsoft.Extensions.Options;
 6using Oracle.ManagedDataAccess.Client;
 7
 8namespace Microsoft.Extensions.DependencyInjection
 9{
 10    /// <summary>DI registration for the Oracle durable-flow state store.</summary>
 11    public static class OracleDurableFlowServiceCollectionExtensions
 12    {
 13        /// <summary>Stores durable-flow state in Oracle Database.</summary>
 14        public static AsyncResponseRegistrationBuilder WithOracleDurableFlows(
 15            this AsyncResponseRegistrationBuilder builder,
 16            Action<OracleDurableFlowOptions>? configure = null)
 17        {
 18            // Singleton on purpose: schema provisioning is cached per store instance, and the
 19            // executor resolves the store from a fresh scope per flow execution — a scoped store
 20            // would re-run EnsureCreated's DDL round-trip on every run.
 21            builder.Services.TryAddSingleton<OracleFlowStateStore>();
 22            return builder.WithDurableFlows<OracleFlowStateStore, OracleDurableFlowOptions>(configure);
 23        }
 24    }
 25}
 26
 27namespace AsyncResponse.DurableFlows.Oracle
 28{
 29/// <summary>Options for the Oracle durable-flow state store.</summary>
 30public sealed class OracleDurableFlowOptions : DurableFlowOptions
 31{
 32    /// <summary>Oracle connection string. Required.</summary>
 33    public string? ConnectionString { get; set; }
 34
 35    /// <summary>Table storing one durable-flow ledger row per flow id.</summary>
 336    public string TableName { get; set; } = "ASYNCRESPONSE_FLOW_STATE";
 37
 38    /// <summary>Creates the table and expiry index on first use.</summary>
 339    public bool AutoCreateSchema { get; set; } = true;
 40
 41    /// <summary>
 42    /// How often <see cref="OracleFlowStateStore.TryCreateAsync"/> opportunistically deletes one bounded
 43    /// batch (1000 rows) of expired rows (loads already treat expired state as absent; pruning
 44    /// bounds table growth). Zero or negative prunes on every save. Default: 5 minutes.
 45    /// </summary>
 346    public TimeSpan PruneInterval { get; set; } = TimeSpan.FromMinutes(5);
 47
 48    /// <summary>
 49    /// Maximum serialized flow-state size in bytes accepted by writes; oversized ledgers fail fast
 50    /// with an actionable error instead of an opaque provider error. Default: <c>null</c>
 51    /// (unlimited — <c>NCLOB</c> is effectively unbounded), settable as an operator budget.
 52    /// </summary>
 53    public long? MaxStateBytes { get; set; }
 54
 55    /// <summary>Validates option values and throws on misconfiguration.</summary>
 56    public void Validate()
 57    {
 358        if (string.IsNullOrWhiteSpace(ConnectionString))
 259            throw new InvalidOperationException($"{nameof(OracleDurableFlowOptions)}.{nameof(ConnectionString)} must be 
 60
 361        DurableFlowStoreShared.ValidateIdentifier(TableName, $"{nameof(OracleDurableFlowOptions)}.{nameof(TableName)}", 
 362        if (MaxStateBytes is <= 0)
 263            throw new InvalidOperationException($"{nameof(OracleDurableFlowOptions)}.{nameof(MaxStateBytes)} must be pos
 364    }
 65}
 66
 67/// <summary>Oracle implementation of <see cref="IFlowStateStore"/>.</summary>
 68public sealed class OracleFlowStateStore : IFlowStateStore
 69{
 70    private const int ObjectAlreadyExists = 955;
 71    private const int ColumnListAlreadyIndexed = 1408;
 72    private const int UniqueConstraintViolated = 1;
 73    private const int PruneBatchSize = 1000;
 74
 75    /// <summary>
 76    /// SQL expression adding a millisecond bind parameter to the database clock. All expiry and
 77    /// lease math runs on <c>SYS_EXTRACT_UTC(SYSTIMESTAMP)</c> so app clock skew can never fence a
 78    /// lease in or out; Oracle NUMBER division keeps fractional seconds, so <c>datetime</c>
 79    /// precision survives the millisecond parameter.
 80    /// </summary>
 81    private static string AddMilliseconds(string parameterName)
 82        => $"SYS_EXTRACT_UTC(SYSTIMESTAMP) + NUMTODSINTERVAL({parameterName} / 1000, 'SECOND')";
 83
 84    private const string UtcNowSql = "SYS_EXTRACT_UTC(SYSTIMESTAMP)";
 85
 86    private readonly OracleDurableFlowOptions _options;
 87    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 88    private long _lastPruneTicks;
 89    private bool _created;
 90
 91    public OracleFlowStateStore(IOptions<OracleDurableFlowOptions> options)
 92    {
 93        _options = options.Value;
 94        _options.Validate();
 95    }
 96
 97    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 98    {
 99        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 100        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 101
 102        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 103        await using var command = connection.CreateCommand();
 104        command.BindByName = true;
 105        command.CommandText = $"SELECT state_json, revision FROM {Table} WHERE flow_id = :flow_id AND expires_at_utc > {
 106        command.Parameters.Add(new OracleParameter("flow_id", flowId));
 107
 108        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 109        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 110            return null;
 111
 112        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 113    }
 114
 115    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 116    {
 117        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 118        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "Oracle");
 119        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 120        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 121            await PruneExpiredAsync(cancellationToken).ConfigureAwait(false);
 122
 123        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 124        try
 125        {
 126            return await TryCreateCoreAsync(connection, flowId, stateJson, state.Revision, ttl, cancellationToken).Confi
 127        }
 128        catch (OracleException ex) when (ex.Number == UniqueConstraintViolated)
 129        {
 130            return await TryCreateCoreAsync(connection, flowId, stateJson, state.Revision, ttl, cancellationToken).Confi
 131        }
 132    }
 133
 134    private async Task<bool> TryCreateCoreAsync(
 135        OracleConnection connection,
 136        string flowId,
 137        string stateJson,
 138        long revision,
 139        TimeSpan ttl,
 140        CancellationToken cancellationToken)
 141    {
 142        await using var command = connection.CreateCommand();
 143        command.BindByName = true;
 144        command.CommandText =
 145            $"""
 146            MERGE INTO {Table} target
 147            USING (SELECT :flow_id AS flow_id FROM dual) source ON (target.flow_id = source.flow_id)
 148            WHEN MATCHED THEN UPDATE SET
 149                target.state_json = :state_json,
 150                target.expires_at_utc = {AddMilliseconds(":ttl_ms")},
 151                target.updated_at_utc = {UtcNowSql},
 152                target.revision = :revision,
 153                target.lease_id = NULL,
 154                target.lease_expires_at_utc = NULL
 155                WHERE target.expires_at_utc <= {UtcNowSql}
 156            WHEN NOT MATCHED THEN
 157                INSERT (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 158                VALUES (:flow_id, :state_json, {AddMilliseconds(":ttl_ms")}, {UtcNowSql}, :revision)
 159            """;
 160        command.Parameters.Add(new OracleParameter("flow_id", flowId));
 161        command.Parameters.Add(new OracleParameter("state_json", OracleDbType.NClob) { Value = stateJson });
 162        command.Parameters.Add(new OracleParameter("ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl)));
 163        command.Parameters.Add(new OracleParameter("revision", revision));
 164        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 165    }
 166
 167    public async Task<bool> TryUpdateAsync(
 168        string flowId,
 169        FlowState state,
 170        long expectedRevision,
 171        TimeSpan ttl,
 172        string? leaseId = null,
 173        CancellationToken cancellationToken = default)
 174    {
 175        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 176        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "Oracle");
 177        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 178
 179        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 180        await using var command = connection.CreateCommand();
 181        command.BindByName = true;
 182        command.CommandText =
 183            $"""
 184            UPDATE {Table}
 185            SET state_json = :state_json,
 186                expires_at_utc = {AddMilliseconds(":ttl_ms")},
 187                updated_at_utc = {UtcNowSql},
 188                revision = :new_revision
 189            WHERE flow_id = :flow_id
 190              AND revision = :expected_revision
 191              AND expires_at_utc > {UtcNowSql}
 192              AND (:lease_id IS NULL OR (lease_id = :lease_id AND lease_expires_at_utc > {UtcNowSql}))
 193            """;
 194        command.Parameters.Add(new OracleParameter("flow_id", flowId));
 195        command.Parameters.Add(new OracleParameter("state_json", OracleDbType.NClob) { Value = stateJson });
 196        command.Parameters.Add(new OracleParameter("ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl)));
 197        command.Parameters.Add(new OracleParameter("expected_revision", expectedRevision));
 198        command.Parameters.Add(new OracleParameter("new_revision", state.Revision));
 199        command.Parameters.Add(new OracleParameter("lease_id", OracleDbType.NVarchar2) { Value = (object?)leaseId ?? DBN
 200        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 201    }
 202
 203    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 204        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 205
 206    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 207        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 208
 209    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 210    {
 211        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 212        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 213        await using var command = connection.CreateCommand();
 214        command.BindByName = true;
 215        command.CommandText = $"UPDATE {Table} SET lease_id = NULL, lease_expires_at_utc = NULL WHERE flow_id = :flow_id
 216        command.Parameters.Add(new OracleParameter("flow_id", flowId));
 217        command.Parameters.Add(new OracleParameter("lease_id", leaseId));
 218        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 219    }
 220
 221    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 222    {
 223        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 224        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 225
 226        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 227        await using var command = connection.CreateCommand();
 228        command.BindByName = true;
 229        command.CommandText = $"DELETE FROM {Table} WHERE flow_id = :flow_id";
 230        command.Parameters.Add(new OracleParameter("flow_id", flowId));
 231        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 232    }
 233
 234    private async Task PruneExpiredAsync(CancellationToken cancellationToken)
 235    {
 236        // One bounded batch per prune interval (policy shared by all relational stores): an
 237        // unbatched DELETE over a large expired backlog holds row locks and bloats one
 238        // transaction for the unlucky create that triggered the prune. Loads already filter on
 239        // expiry, so any backlog beyond the batch just waits for the next interval.
 240        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 241        await using var command = connection.CreateCommand();
 242        command.BindByName = true;
 243        command.CommandText = $"DELETE FROM {Table} WHERE expires_at_utc <= {UtcNowSql} AND ROWNUM <= {PruneBatchSize}";
 244        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 245    }
 246
 247    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 248    {
 249        if (_created || !_options.AutoCreateSchema)
 250            return;
 251
 252        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 253        try
 254        {
 255            if (_created)
 256                return;
 257
 258            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 259            await ExecuteIgnoringExistsAsync(
 260                connection,
 261                $"""
 262                CREATE TABLE {Table} (
 263                    flow_id NVARCHAR2(400) NOT NULL PRIMARY KEY,
 264                    state_json NCLOB NOT NULL,
 265                    expires_at_utc TIMESTAMP(6) NOT NULL,
 266                    updated_at_utc TIMESTAMP(6) NOT NULL,
 267                    revision NUMBER(19) DEFAULT 0 NOT NULL,
 268                    lease_id NVARCHAR2(64) NULL,
 269                    lease_expires_at_utc TIMESTAMP(6) NULL
 270                )
 271                """,
 272                cancellationToken).ConfigureAwait(false);
 273            await ExecuteIgnoringExistsAsync(
 274                connection,
 275                $"CREATE INDEX {IndexName} ON {Table} (expires_at_utc)",
 276                cancellationToken).ConfigureAwait(false);
 277            _created = true;
 278        }
 279        finally
 280        {
 281            _ensureGate.Release();
 282        }
 283    }
 284
 285    private static async Task ExecuteIgnoringExistsAsync(OracleConnection connection, string commandText, CancellationTo
 286    {
 287        await using var command = connection.CreateCommand();
 288        command.CommandText = commandText;
 289        try
 290        {
 291            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 292        }
 293        catch (OracleException ex) when (ex.Number is ObjectAlreadyExists or ColumnListAlreadyIndexed)
 294        {
 295            // ORA-00955: the object (table/index name) already exists. ORA-01408: the column list
 296            // is already indexed — raised instead of ORA-00955 when an operator pre-created the
 297            // expiry index under a different name; the index we want exists in substance.
 298        }
 299    }
 300
 301    private async Task<bool> UpdateLeaseAsync(
 302        string flowId,
 303        string leaseId,
 304        TimeSpan leaseDuration,
 305        bool acquire,
 306        CancellationToken cancellationToken)
 307    {
 308        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 309        ArgumentException.ThrowIfNullOrWhiteSpace(leaseId);
 310        if (leaseDuration <= TimeSpan.Zero)
 311            throw new ArgumentOutOfRangeException(nameof(leaseDuration));
 312
 313        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 314        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 315        await using var command = connection.CreateCommand();
 316        command.BindByName = true;
 317        // Lease fencing runs entirely on the database clock: acquire steals only leases the
 318        // database considers expired, and renew/extend stays relative to the server's UTC time,
 319        // so worker clock skew can never make two nodes hold the same lease.
 320        command.CommandText =
 321            $"""
 322            UPDATE {Table}
 323            SET lease_id = :lease_id, lease_expires_at_utc = {AddMilliseconds(":lease_ms")}
 324            WHERE flow_id = :flow_id
 325              AND expires_at_utc > {UtcNowSql}
 326              AND {(acquire ? $"(lease_id IS NULL OR lease_expires_at_utc <= {UtcNowSql} OR lease_id = :lease_id)" : $"l
 327            """;
 328        command.Parameters.Add(new OracleParameter("flow_id", flowId));
 329        command.Parameters.Add(new OracleParameter("lease_id", leaseId));
 330        command.Parameters.Add(new OracleParameter("lease_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(leaseDu
 331        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 332    }
 333
 334    private async Task<OracleConnection> OpenConnectionAsync(CancellationToken cancellationToken)
 335    {
 336        var connection = new OracleConnection(_options.ConnectionString);
 337        try
 338        {
 339            await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
 340            return connection;
 341        }
 342        catch
 343        {
 344            await connection.DisposeAsync().ConfigureAwait(false);
 345            throw;
 346        }
 347    }
 348
 349    private string Table => _options.TableName;
 350    private string IndexName => $"{_options.TableName}_EXPIRES_IDX";
 351}
 352}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor()
Validate()