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

Information
Class: AsyncResponse.DurableFlows.Oracle.OracleFlowStateStore
Assembly: AsyncResponse.DurableFlows.Oracle
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.Oracle/OracleDurableFlows.cs
Line coverage
98%
Covered lines: 167
Uncovered lines: 2
Coverable lines: 169
Total lines: 352
Line coverage: 98.8%
Branch coverage
81%
Covered branches: 13
Total branches: 16
Branch coverage: 81.2%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
AddMilliseconds(...)100%11100%
.ctor(...)100%11100%
LoadAsync()100%11100%
TryCreateAsync()100%2280%
TryCreateCoreAsync()100%11100%
TryUpdateAsync()100%22100%
TryAcquireLeaseAsync(...)100%11100%
TryRenewLeaseAsync(...)100%11100%
ReleaseLeaseAsync()100%11100%
TryDeleteAsync()100%11100%
PruneExpiredAsync()100%11100%
EnsureCreatedAsync()83.33%6696.43%
ExecuteIgnoringExistsAsync()100%11100%
UpdateLeaseAsync()75%4495.24%
OpenConnectionAsync()50%2271.43%
get_Table()100%11100%
get_IndexName()100%11100%

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>
 36    public string TableName { get; set; } = "ASYNCRESPONSE_FLOW_STATE";
 37
 38    /// <summary>Creates the table and expiry index on first use.</summary>
 39    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>
 46    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    {
 58        if (string.IsNullOrWhiteSpace(ConnectionString))
 59            throw new InvalidOperationException($"{nameof(OracleDurableFlowOptions)}.{nameof(ConnectionString)} must be 
 60
 61        DurableFlowStoreShared.ValidateIdentifier(TableName, $"{nameof(OracleDurableFlowOptions)}.{nameof(TableName)}", 
 62        if (MaxStateBytes is <= 0)
 63            throw new InvalidOperationException($"{nameof(OracleDurableFlowOptions)}.{nameof(MaxStateBytes)} must be pos
 64    }
 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)
 182        => $"SYS_EXTRACT_UTC(SYSTIMESTAMP) + NUMTODSINTERVAL({parameterName} / 1000, 'SECOND')";
 83
 84    private const string UtcNowSql = "SYS_EXTRACT_UTC(SYSTIMESTAMP)";
 85
 86    private readonly OracleDurableFlowOptions _options;
 387    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 88    private long _lastPruneTicks;
 89    private bool _created;
 90
 391    public OracleFlowStateStore(IOptions<OracleDurableFlowOptions> options)
 92    {
 393        _options = options.Value;
 394        _options.Validate();
 395    }
 96
 97    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 98    {
 399        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 3100        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 101
 3102        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1103        await using var command = connection.CreateCommand();
 1104        command.BindByName = true;
 1105        command.CommandText = $"SELECT state_json, revision FROM {Table} WHERE flow_id = :flow_id AND expires_at_utc > {
 1106        command.Parameters.Add(new OracleParameter("flow_id", flowId));
 107
 1108        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 1109        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 1110            return null;
 111
 1112        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 1113    }
 114
 115    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 116    {
 1117        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 1118        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "Oracle");
 1119        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1120        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 1121            await PruneExpiredAsync(cancellationToken).ConfigureAwait(false);
 122
 1123        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 124        try
 125        {
 1126            return await TryCreateCoreAsync(connection, flowId, stateJson, state.Revision, ttl, cancellationToken).Confi
 127        }
 0128        catch (OracleException ex) when (ex.Number == UniqueConstraintViolated)
 129        {
 0130            return await TryCreateCoreAsync(connection, flowId, stateJson, state.Revision, ttl, cancellationToken).Confi
 131        }
 1132    }
 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    {
 1142        await using var command = connection.CreateCommand();
 1143        command.BindByName = true;
 1144        command.CommandText =
 1145            $"""
 1146            MERGE INTO {Table} target
 1147            USING (SELECT :flow_id AS flow_id FROM dual) source ON (target.flow_id = source.flow_id)
 1148            WHEN MATCHED THEN UPDATE SET
 1149                target.state_json = :state_json,
 1150                target.expires_at_utc = {AddMilliseconds(":ttl_ms")},
 1151                target.updated_at_utc = {UtcNowSql},
 1152                target.revision = :revision,
 1153                target.lease_id = NULL,
 1154                target.lease_expires_at_utc = NULL
 1155                WHERE target.expires_at_utc <= {UtcNowSql}
 1156            WHEN NOT MATCHED THEN
 1157                INSERT (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 1158                VALUES (:flow_id, :state_json, {AddMilliseconds(":ttl_ms")}, {UtcNowSql}, :revision)
 1159            """;
 1160        command.Parameters.Add(new OracleParameter("flow_id", flowId));
 1161        command.Parameters.Add(new OracleParameter("state_json", OracleDbType.NClob) { Value = stateJson });
 1162        command.Parameters.Add(new OracleParameter("ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl)));
 1163        command.Parameters.Add(new OracleParameter("revision", revision));
 1164        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 1165    }
 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    {
 1175        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 1176        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "Oracle");
 1177        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 178
 1179        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1180        await using var command = connection.CreateCommand();
 1181        command.BindByName = true;
 1182        command.CommandText =
 1183            $"""
 1184            UPDATE {Table}
 1185            SET state_json = :state_json,
 1186                expires_at_utc = {AddMilliseconds(":ttl_ms")},
 1187                updated_at_utc = {UtcNowSql},
 1188                revision = :new_revision
 1189            WHERE flow_id = :flow_id
 1190              AND revision = :expected_revision
 1191              AND expires_at_utc > {UtcNowSql}
 1192              AND (:lease_id IS NULL OR (lease_id = :lease_id AND lease_expires_at_utc > {UtcNowSql}))
 1193            """;
 1194        command.Parameters.Add(new OracleParameter("flow_id", flowId));
 1195        command.Parameters.Add(new OracleParameter("state_json", OracleDbType.NClob) { Value = stateJson });
 1196        command.Parameters.Add(new OracleParameter("ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl)));
 1197        command.Parameters.Add(new OracleParameter("expected_revision", expectedRevision));
 1198        command.Parameters.Add(new OracleParameter("new_revision", state.Revision));
 1199        command.Parameters.Add(new OracleParameter("lease_id", OracleDbType.NVarchar2) { Value = (object?)leaseId ?? DBN
 1200        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 1201    }
 202
 203    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 3204        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 205
 206    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 1207        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 208
 209    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 210    {
 1211        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1212        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1213        await using var command = connection.CreateCommand();
 1214        command.BindByName = true;
 1215        command.CommandText = $"UPDATE {Table} SET lease_id = NULL, lease_expires_at_utc = NULL WHERE flow_id = :flow_id
 1216        command.Parameters.Add(new OracleParameter("flow_id", flowId));
 1217        command.Parameters.Add(new OracleParameter("lease_id", leaseId));
 1218        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1219    }
 220
 221    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 222    {
 1223        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 1224        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 225
 1226        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1227        await using var command = connection.CreateCommand();
 1228        command.BindByName = true;
 1229        command.CommandText = $"DELETE FROM {Table} WHERE flow_id = :flow_id";
 1230        command.Parameters.Add(new OracleParameter("flow_id", flowId));
 1231        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 1232    }
 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.
 1240        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1241        await using var command = connection.CreateCommand();
 1242        command.BindByName = true;
 1243        command.CommandText = $"DELETE FROM {Table} WHERE expires_at_utc <= {UtcNowSql} AND ROWNUM <= {PruneBatchSize}";
 1244        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1245    }
 246
 247    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 248    {
 3249        if (_created || !_options.AutoCreateSchema)
 3250            return;
 251
 3252        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 253        try
 254        {
 3255            if (_created)
 2256                return;
 257
 1258            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1259            await ExecuteIgnoringExistsAsync(
 1260                connection,
 1261                $"""
 1262                CREATE TABLE {Table} (
 1263                    flow_id NVARCHAR2(400) NOT NULL PRIMARY KEY,
 1264                    state_json NCLOB NOT NULL,
 1265                    expires_at_utc TIMESTAMP(6) NOT NULL,
 1266                    updated_at_utc TIMESTAMP(6) NOT NULL,
 1267                    revision NUMBER(19) DEFAULT 0 NOT NULL,
 1268                    lease_id NVARCHAR2(64) NULL,
 1269                    lease_expires_at_utc TIMESTAMP(6) NULL
 1270                )
 1271                """,
 1272                cancellationToken).ConfigureAwait(false);
 1273            await ExecuteIgnoringExistsAsync(
 1274                connection,
 1275                $"CREATE INDEX {IndexName} ON {Table} (expires_at_utc)",
 1276                cancellationToken).ConfigureAwait(false);
 1277            _created = true;
 1278        }
 279        finally
 280        {
 3281            _ensureGate.Release();
 282        }
 3283    }
 284
 285    private static async Task ExecuteIgnoringExistsAsync(OracleConnection connection, string commandText, CancellationTo
 286    {
 1287        await using var command = connection.CreateCommand();
 1288        command.CommandText = commandText;
 289        try
 290        {
 1291            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1292        }
 1293        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.
 1298        }
 1299    }
 300
 301    private async Task<bool> UpdateLeaseAsync(
 302        string flowId,
 303        string leaseId,
 304        TimeSpan leaseDuration,
 305        bool acquire,
 306        CancellationToken cancellationToken)
 307    {
 3308        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 3309        ArgumentException.ThrowIfNullOrWhiteSpace(leaseId);
 3310        if (leaseDuration <= TimeSpan.Zero)
 2311            throw new ArgumentOutOfRangeException(nameof(leaseDuration));
 312
 1313        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1314        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1315        await using var command = connection.CreateCommand();
 1316        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.
 1320        command.CommandText =
 1321            $"""
 1322            UPDATE {Table}
 1323            SET lease_id = :lease_id, lease_expires_at_utc = {AddMilliseconds(":lease_ms")}
 1324            WHERE flow_id = :flow_id
 1325              AND expires_at_utc > {UtcNowSql}
 1326              AND {(acquire ? $"(lease_id IS NULL OR lease_expires_at_utc <= {UtcNowSql} OR lease_id = :lease_id)" : $"l
 1327            """;
 1328        command.Parameters.Add(new OracleParameter("flow_id", flowId));
 1329        command.Parameters.Add(new OracleParameter("lease_id", leaseId));
 1330        command.Parameters.Add(new OracleParameter("lease_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(leaseDu
 1331        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 1332    }
 333
 334    private async Task<OracleConnection> OpenConnectionAsync(CancellationToken cancellationToken)
 335    {
 3336        var connection = new OracleConnection(_options.ConnectionString);
 337        try
 338        {
 3339            await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
 1340            return connection;
 341        }
 2342        catch
 343        {
 2344            await connection.DisposeAsync().ConfigureAwait(false);
 2345            throw;
 346        }
 1347    }
 348
 1349    private string Table => _options.TableName;
 1350    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

AddMilliseconds(string)
.ctor(Microsoft.Extensions.Options.IOptions<AsyncResponse.DurableFlows.Oracle.OracleDurableFlowOptions>)
LoadAsync()
TryCreateAsync()
TryCreateCoreAsync()
TryUpdateAsync()
TryAcquireLeaseAsync(string, string, System.TimeSpan, System.Threading.CancellationToken)
TryRenewLeaseAsync(string, string, System.TimeSpan, System.Threading.CancellationToken)
ReleaseLeaseAsync()
TryDeleteAsync()
PruneExpiredAsync()
EnsureCreatedAsync()
ExecuteIgnoringExistsAsync()
UpdateLeaseAsync()
OpenConnectionAsync()
get_Table()
get_IndexName()