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

Information
Class: AsyncResponse.DurableFlows.Oracle.OracleFlowStateStore
Assembly: AsyncResponse.DurableFlows.Oracle
File(s): /_/src/DurableFlows/AsyncResponse.DurableFlows.Oracle/OracleDurableFlows.cs
Line coverage
96%
Covered lines: 373
Uncovered lines: 15
Coverable lines: 388
Total lines: 851
Line coverage: 96.1%
Branch coverage
86%
Covered branches: 104
Total branches: 120
Branch coverage: 86.6%
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%44100%
ValidateCreate(...)50%22100%
TryCreateAsync()100%22100%
TryCreateCoreAsync()100%11100%
TryUpdateAsync()100%11100%
TryAcquireLeaseAsync(...)100%11100%
TryRenewLeaseAsync(...)100%11100%
ReleaseLeaseAsync()100%11100%
ObserveLeaseAsync()100%88100%
TryDeleteAsync()100%11100%
PruneExpiredAsync()100%11100%
EnsureCreatedAsync()100%88100%
NationalId(...)100%22100%
ExecuteIgnoringExistsAsync()100%11100%
VerifyFlowTableAsync()78.57%292887.75%
ResolveBaseTableAsync()75%171683.72%
ResolveSynonymAsync()66.66%6688.88%
VerifyComparisonSemanticsAsync()100%66100%
IsBinary()100%11100%
DiagnoseComparisonSemantics(...)100%88100%
get_DataType()100%11100%
get_Nullable()100%11100%
get_CharLength()100%11100%
get_Precision()100%11100%
get_Scale()100%11100%
get_HasDefault()100%11100%
get_Virtual()100%11100%
get_Identity()100%11100%
get_IsWritableWithoutValue()100%66100%
get_Name()100%11100%
Mismatch(...)88.46%2626100%
.cctor()100%11100%
VerifyFlowIdIsUniqueAsync()100%22100%
UpdateLeaseAsync()100%22100%
OpenConnectionAsync(...)100%11100%
get_Table()100%11100%
get_IndexName()100%11100%
get_CatalogName()100%11100%

File(s)

/_/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.Logging;
 6using Microsoft.Extensions.Options;
 7using Oracle.ManagedDataAccess.Client;
 8
 9namespace Microsoft.Extensions.DependencyInjection
 10{
 11    /// <summary>DI registration for the Oracle durable-flow state store.</summary>
 12    public static class OracleDurableFlowServiceCollectionExtensions
 13    {
 14        /// <summary>Stores durable-flow state in Oracle Database.</summary>
 15        public static AsyncResponseRegistrationBuilder WithOracleDurableFlows(
 16            this AsyncResponseRegistrationBuilder builder,
 17            Action<OracleDurableFlowOptions>? configure = null)
 18        {
 19            // Singleton on purpose: schema provisioning is cached per store instance, and the
 20            // executor resolves the store from a fresh scope per flow execution — a scoped store
 21            // would re-run EnsureCreated's DDL round-trip on every run.
 22            builder.Services.TryAddSingleton<OracleFlowStateStore>();
 23            return builder.WithDurableFlows<OracleFlowStateStore, OracleDurableFlowOptions>(configure);
 24        }
 25    }
 26}
 27
 28namespace AsyncResponse.DurableFlows.Oracle
 29{
 30/// <summary>Options for the Oracle durable-flow state store.</summary>
 31public sealed class OracleDurableFlowOptions : DurableFlowOptions
 32{
 33    /// <summary>Oracle connection string. Required.</summary>
 34    public string? ConnectionString { get; set; }
 35
 36    /// <summary>Table storing one durable-flow ledger row per flow id.</summary>
 37    public string TableName { get; set; } = "ASYNCRESPONSE_FLOW_STATE";
 38
 39    /// <summary>Creates the table and expiry index on first use.</summary>
 40    public bool AutoCreateSchema { get; set; } = true;
 41
 42    /// <summary>
 43    /// How often <see cref="OracleFlowStateStore.TryCreateAsync"/> opportunistically deletes one bounded
 44    /// batch (1000 rows) of expired rows (loads already treat expired state as absent; pruning
 45    /// bounds table growth). Zero or negative prunes on every save. Default: 5 minutes.
 46    /// </summary>
 47    public TimeSpan PruneInterval { get; set; } = TimeSpan.FromMinutes(5);
 48
 49    /// <summary>
 50    /// Wall-clock budget one opportunistic prune may spend draining expired rows in batches of
 51    /// 1000 after its first batch (the first always runs). A single batch per interval capped
 52    /// cleanup at ~3 rows/second, which any busier instance outgrew forever; the prune now drains
 53    /// batches until one comes back short or this budget lapses, and reports the outcome on the
 54    /// <c>AsyncResponse</c> meter (<c>asyncresponse.flow_state.pruned_rows</c>,
 55    /// <c>prune_failures</c>, <c>prune_budget_exhausted</c>) and the store's logger. The create
 56    /// that triggers the prune waits for it, so this bounds that create's added latency. Zero
 57    /// keeps the historical single batch. Default: 2 seconds.
 58    /// </summary>
 59    public TimeSpan PruneBudget { get; set; } = DurableFlowStoreShared.DefaultPruneBudget;
 60
 61    /// <summary>
 62    /// Maximum serialized flow-state size in bytes accepted by writes; oversized ledgers fail fast
 63    /// with an actionable error instead of an opaque provider error. Default: <c>null</c>
 64    /// (unlimited — <c>NCLOB</c> is effectively unbounded), settable as an operator budget.
 65    /// </summary>
 66    public long? MaxStateBytes { get; set; }
 67
 68    /// <summary>Validates option values and throws on misconfiguration.</summary>
 69    public void Validate()
 70    {
 71        DurableFlowStoreShared.ValidateConnectionString(ConnectionString, nameof(OracleDurableFlowOptions));
 72        DurableFlowStoreShared.ValidateIdentifier(TableName, $"{nameof(OracleDurableFlowOptions)}.{nameof(TableName)}", 
 73
 74        // Indexes share Oracle's schema-object namespace with tables: a table whose name ends
 75        // exactly where the reserved "_EXPIRES_IDX" stem truncates derives its own name, and the
 76        // CREATE INDEX then raises ORA-00955 — indistinguishable from the benign already-exists
 77        // race the DDL path deliberately swallows — so the expiry index would silently never
 78        // exist and every prune would full-scan. Unquoted identifiers are case-insensitive.
 79        if (string.Equals(DurableFlowStoreShared.DerivedName(TableName, "_EXPIRES_IDX", 128), TableName, StringCompariso
 80            throw new InvalidOperationException(
 81                $"{nameof(OracleDurableFlowOptions)}.{nameof(TableName)} '{TableName}' collides with its derived expiry-
 82        DurableFlowStoreShared.ValidateMaxStateBytes(MaxStateBytes, nameof(OracleDurableFlowOptions));
 83        DurableFlowStoreShared.ValidatePruneBudget(PruneBudget, nameof(OracleDurableFlowOptions));
 84    }
 85}
 86
 87/// <summary>Oracle implementation of <see cref="IFlowStateStore"/>.</summary>
 88public sealed class OracleFlowStateStore : IFlowStateStore
 89{
 90    private const int ObjectAlreadyExists = 955;
 91    private const int ColumnListAlreadyIndexed = 1408;
 92    private const int UniqueConstraintViolated = 1;
 93    private readonly ILogger<OracleFlowStateStore>? _logger;
 94
 95    /// <summary>
 96    /// SQL expression adding a millisecond bind parameter to the database clock. All expiry and
 97    /// lease math runs on <c>SYS_EXTRACT_UTC(SYSTIMESTAMP)</c> so app clock skew can never fence a
 98    /// lease in or out; Oracle NUMBER division keeps fractional seconds, so <c>datetime</c>
 99    /// precision survives the millisecond parameter.
 100    /// </summary>
 101    private static string AddMilliseconds(string parameterName)
 1718102        => $"SYS_EXTRACT_UTC(SYSTIMESTAMP) + NUMTODSINTERVAL({parameterName} / 1000, 'SECOND')";
 103
 104    private const string UtcNowSql = "SYS_EXTRACT_UTC(SYSTIMESTAMP)";
 105
 106    private readonly OracleDurableFlowOptions _options;
 222107    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 108    private long _lastPruneTicks;
 109    private volatile bool _created;
 110
 222111    public OracleFlowStateStore(IOptions<OracleDurableFlowOptions> options, ILogger<OracleFlowStateStore>? logger = null
 112    {
 222113        _logger = logger;
 222114        _options = options.Value;
 222115        _options.Validate();
 222116    }
 117
 118    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 119    {
 682120        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 682121        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 122
 680123        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 678124        await using var command = connection.CreateCommand();
 678125        command.BindByName = true;
 678126        command.CommandText = $"SELECT state_json, revision FROM {Table} WHERE flow_id = :flow_id AND expires_at_utc > {
 678127        command.Parameters.Add(NationalId("flow_id", flowId));
 128
 678129        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 678130        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 5131            return null;
 132
 673133        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 677134    }
 135
 136    /// <inheritdoc />
 137    public void ValidateCreate(string flowId, FlowState state, TimeSpan ttl)
 138    {
 136139        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 136140        if (_options.MaxStateBytes is not null)
 4141            _ = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "Oracle");
 134142    }
 143
 144    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 145    {
 305146        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 304147        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "Oracle");
 304148        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 298149        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 278150            await DurableFlowStoreShared.PruneQuietlyAsync(() => PruneExpiredAsync(cancellationToken), _options.PruneBud
 151
 298152        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 153        try
 154        {
 298155            return await TryCreateCoreAsync(connection, flowId, stateJson, state.Revision, ttl, cancellationToken).Confi
 156        }
 43157        catch (OracleException ex) when (ex.Number == UniqueConstraintViolated)
 158        {
 42159            return await TryCreateCoreAsync(connection, flowId, stateJson, state.Revision, ttl, cancellationToken).Confi
 160        }
 297161    }
 162
 163    private async Task<bool> TryCreateCoreAsync(
 164        OracleConnection connection,
 165        string flowId,
 166        string stateJson,
 167        long revision,
 168        TimeSpan ttl,
 169        CancellationToken cancellationToken)
 170    {
 340171        await using var command = connection.CreateCommand();
 340172        command.BindByName = true;
 340173        command.CommandText =
 340174            $"""
 340175            MERGE INTO {Table} target
 340176            USING (SELECT :flow_id AS flow_id FROM dual) source ON (target.flow_id = source.flow_id)
 340177            WHEN MATCHED THEN UPDATE SET
 340178                target.state_json = :state_json,
 340179                target.expires_at_utc = {AddMilliseconds(":ttl_ms")},
 340180                target.updated_at_utc = {UtcNowSql},
 340181                target.revision = :revision,
 340182                target.lease_id = NULL,
 340183                target.lease_expires_at_utc = NULL
 340184                WHERE target.expires_at_utc <= {UtcNowSql}
 340185            WHEN NOT MATCHED THEN
 340186                INSERT (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 340187                VALUES (:flow_id, :state_json, {AddMilliseconds(":ttl_ms")}, {UtcNowSql}, :revision)
 340188            """;
 340189        command.Parameters.Add(NationalId("flow_id", flowId));
 340190        command.Parameters.Add(new OracleParameter("state_json", OracleDbType.NClob) { Value = stateJson });
 340191        command.Parameters.Add(new OracleParameter("ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl)));
 340192        command.Parameters.Add(new OracleParameter("revision", revision));
 340193        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 297194    }
 195
 196    public async Task<bool> TryUpdateAsync(
 197        string flowId,
 198        FlowState state,
 199        long expectedRevision,
 200        TimeSpan ttl,
 201        string? leaseId = null,
 202        CancellationToken cancellationToken = default)
 203    {
 875204        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 875205        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "Oracle");
 875206        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 207
 875208        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 875209        await using var command = connection.CreateCommand();
 875210        command.BindByName = true;
 875211        command.CommandText =
 875212            $"""
 875213            UPDATE {Table}
 875214            SET state_json = :state_json,
 875215                expires_at_utc = {AddMilliseconds(":ttl_ms")},
 875216                updated_at_utc = {UtcNowSql},
 875217                revision = :new_revision
 875218            WHERE flow_id = :flow_id
 875219              AND revision = :expected_revision
 875220              AND expires_at_utc > {UtcNowSql}
 875221              AND (:lease_id IS NULL OR (lease_id = :lease_id AND lease_expires_at_utc > {UtcNowSql}))
 875222            """;
 875223        command.Parameters.Add(NationalId("flow_id", flowId));
 875224        command.Parameters.Add(new OracleParameter("state_json", OracleDbType.NClob) { Value = stateJson });
 875225        command.Parameters.Add(new OracleParameter("ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl)));
 875226        command.Parameters.Add(new OracleParameter("expected_revision", expectedRevision));
 875227        command.Parameters.Add(new OracleParameter("new_revision", state.Revision));
 875228        command.Parameters.Add(NationalId("lease_id", leaseId));
 875229        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 875230    }
 231
 232    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 153233        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 234
 235    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 12236        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 237
 238    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 239    {
 145240        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 145241        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 145242        await using var command = connection.CreateCommand();
 145243        command.BindByName = true;
 145244        command.CommandText = $"UPDATE {Table} SET lease_id = NULL, lease_expires_at_utc = NULL WHERE flow_id = :flow_id
 145245        command.Parameters.Add(NationalId("flow_id", flowId));
 145246        command.Parameters.Add(NationalId("lease_id", leaseId));
 145247        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 144248    }
 249
 250    /// <inheritdoc />
 251    public async Task<FlowLeaseObservation?> ObserveLeaseAsync(string flowId, CancellationToken cancellationToken = defa
 252    {
 18253        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 12254        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 255
 256        // The two lease columns exactly as stored — deliberately no SYS_EXTRACT_UTC(SYSTIMESTAMP)
 257        // predicate, unlike every other statement in this store: an expired lease nobody has taken
 258        // over must keep reading as the same lease, because the engine's proof of a live holder is
 259        // that two observations DIFFER. Whether it has lapsed stays UpdateLeaseAsync's call, on the
 260        // database clock. TIMESTAMP(6) carries no zone and reads back Unspecified; the value is
 261        // SYS_EXTRACT_UTC arithmetic, and the shared shaper stamps it UTC.
 12262        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 12263        await using var command = connection.CreateCommand();
 12264        command.BindByName = true;
 12265        command.CommandText = $"SELECT lease_id, lease_expires_at_utc FROM {Table} WHERE flow_id = :flow_id";
 12266        command.Parameters.Add(NationalId("flow_id", flowId));
 267
 12268        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 12269        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 2270            return FlowLeaseObservation.Unheld;
 271
 10272        return DurableFlowStoreShared.LeaseObservation(
 10273            reader.IsDBNull(0) ? null : reader.GetString(0),
 10274            reader.IsDBNull(1) ? null : reader.GetDateTime(1));
 12275    }
 276
 277    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 278    {
 10279        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 10280        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 281
 10282        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 10283        await using var command = connection.CreateCommand();
 10284        command.BindByName = true;
 10285        command.CommandText = $"DELETE FROM {Table} WHERE flow_id = :flow_id";
 10286        command.Parameters.Add(NationalId("flow_id", flowId));
 10287        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 10288    }
 289
 290    private async Task<int> PruneExpiredAsync(CancellationToken cancellationToken)
 291    {
 292        // One bounded batch per call; DurableFlowStoreShared.PruneQuietlyAsync repeats it under
 293        // the PruneBudget while batches come back full (policy shared by all relational stores): an
 294        // unbatched DELETE over a large expired backlog holds row locks and bloats one
 295        // transaction for the unlucky create that triggered the prune. Loads already filter on
 296        // expiry, so any backlog beyond the batch just waits for the next interval.
 139297        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 139298        await using var command = connection.CreateCommand();
 139299        command.BindByName = true;
 139300        command.CommandText = $"DELETE FROM {Table} WHERE expires_at_utc <= {UtcNowSql} AND ROWNUM <= {DurableFlowStoreS
 139301        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 138302    }
 303
 304    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 305    {
 2191306        if (_created)
 1937307            return;
 308
 254309        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 310        try
 311        {
 254312            if (_created)
 106313                return;
 314
 148315            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 146316            if (_options.AutoCreateSchema)
 317            {
 137318                await ExecuteIgnoringExistsAsync(
 137319                    connection,
 137320                    $"""
 137321                    CREATE TABLE {Table} (
 137322                        flow_id NVARCHAR2(400) NOT NULL PRIMARY KEY,
 137323                        state_json NCLOB NOT NULL,
 137324                        expires_at_utc TIMESTAMP(6) NOT NULL,
 137325                        updated_at_utc TIMESTAMP(6) NOT NULL,
 137326                        revision NUMBER(19) DEFAULT 0 NOT NULL,
 137327                        lease_id NVARCHAR2(64) NULL,
 137328                        lease_expires_at_utc TIMESTAMP(6) NULL
 137329                    )
 137330                    """,
 137331                    cancellationToken).ConfigureAwait(false);
 137332                await ExecuteIgnoringExistsAsync(
 137333                    connection,
 137334                    $"CREATE INDEX {IndexName} ON {Table} (expires_at_utc)",
 137335                    cancellationToken).ConfigureAwait(false);
 336            }
 337
 338            // Oracle DDL commits implicitly, so everything above is already committed and the
 339            // checks below run outside any transaction: they read the catalog for OTHER sessions'
 340            // committed objects and must never sit on DDL locks of their own. _created latches
 341            // only on a VERIFIED table: when the table does not exist yet (AutoCreateSchema =
 342            // false, migration not run), verification re-runs on the next operation instead of
 343            // being silently skipped for the process lifetime.
 146344            _created = await VerifyFlowTableAsync(connection, cancellationToken).ConfigureAwait(false);
 140345        }
 346        finally
 347        {
 254348            _ensureGate.Release();
 349        }
 2183350    }
 351
 352    // NVarchar2, never the inferred Varchar2: ids travel to NVARCHAR2 columns, and a Varchar2
 353    // bind converts through the DATABASE character set on the wire — on a non-Unicode
 354    // NLS_CHARACTERSET (e.g. WE8MSWIN1252) two distinct Unicode ids collapse to one '???' key,
 355    // so one flow's row answers another flow's create/load. The verifier enforces NVARCHAR2
 356    // columns for exactly this reason; the binds must match it. Internal (not private) so the
 357    // unit suite can pin the bind type without an Oracle server.
 358    internal static OracleParameter NationalId(string name, string? value)
 3410359        => new(name, OracleDbType.NVarchar2) { Value = (object?)value ?? DBNull.Value };
 360
 361    private static async Task ExecuteIgnoringExistsAsync(OracleConnection connection, string commandText, CancellationTo
 362    {
 274363        await using var command = connection.CreateCommand();
 274364        command.CommandText = commandText;
 365        try
 366        {
 274367            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 270368        }
 4369        catch (OracleException ex) when (ex.Number is ObjectAlreadyExists or ColumnListAlreadyIndexed)
 370        {
 371            // ORA-00955: the object (table/index name) already exists. ORA-01408: the column list
 372            // is already indexed — raised instead of ORA-00955 when an operator pre-created the
 373            // expiry index under a different name; the index we want exists in substance.
 4374        }
 274375    }
 376
 377    /// <summary>
 378    /// Checks the objects this store will actually use, independently of who created them.
 379    /// <see cref="ExecuteIgnoringExistsAsync"/> swallows ORA-00955, so a pre-existing object under
 380    /// the table's name — an earlier build's table, a hand-written one, even a VIEW — is left
 381    /// exactly as it was, and <c>AutoCreateSchema = false</c> issues no DDL at all: the CREATE
 382    /// above only ever protects a table this build created. Three properties of what the name
 383    /// resolves to are load-bearing and all fail SILENTLY (or mid-operation) when absent, which is
 384    /// why they are checked once per store instance rather than left to the first query:
 385    /// <list type="bullet">
 386    /// <item><description>
 387    /// Ordinal comparison in this session. NLS_COMP=LINGUISTIC plus a case-insensitive NLS_SORT
 388    /// folds every <c>flow_id = :flow_id</c> predicate this store runs while the primary-key index
 389    /// stays binary and admits both casings — loads and leases silently cross two flows. Checked
 390    /// first and always, even before the table exists: it poisons every future query no matter who
 391    /// creates the table.
 392    /// </description></item>
 393    /// <item><description>
 394    /// A real TABLE under the name. The column views below happily describe a VIEW, which has no
 395    /// key to raise ORA-00001, so it would pass every shape check and then lose duplicate
 396    /// detection (or fail outright) at the first MERGE.
 397    /// </description></item>
 398    /// <item><description>
 399    /// A unique key on flow_id alone. <see cref="TryCreateAsync"/> is the engine's insert-if-absent
 400    /// primitive: its MERGE detects "already exists" from ORA-00001 when two creates race — with no
 401    /// such key nothing raises it, both MERGEs insert, and one flow id gets two rows and two
 402    /// executions.
 403    /// </description></item>
 404    /// </list>
 405    /// The expiry index is deliberately not verified: it is performance-only (loads and pruning
 406    /// filter on expiry either way), the same standard the sibling stores apply to theirs.
 407    /// </summary>
 408    private async Task<bool> VerifyFlowTableAsync(OracleConnection connection, CancellationToken cancellationToken)
 409    {
 146410        await VerifyComparisonSemanticsAsync(connection, cancellationToken).ConfigureAwait(false);
 411
 146412        if (await ResolveBaseTableAsync(connection, cancellationToken).ConfigureAwait(false) is not { } table)
 413        {
 414            // The table does not exist: AutoCreateSchema = false and the migration has not run yet.
 415            // That surfaces at the first query with a clear ORA-00942, and failing here would break
 416            // the documented "create it yourself, later" workflow. Returning false leaves _created
 417            // unlatched, so a table created later is still verified before it is trusted.
 1418            return false;
 419        }
 420
 144421        var columns = new Dictionary<string, ActualColumn>(StringComparer.OrdinalIgnoreCase);
 144422        await using (var command = connection.CreateCommand())
 423        {
 144424            command.BindByName = true;
 425            // ALL_TAB_COLS rather than ALL_TAB_COLUMNS: it carries VIRTUAL_COLUMN and
 426            // IDENTITY_COLUMN, and HIDDEN_COLUMN = 'NO' keeps user columns — including INVISIBLE
 427            // ones, which still break INSERTs that do not name them — while dropping the
 428            // system-generated ones function-based indexes add. DEFAULT_LENGTH stands in for
 429            // DATA_DEFAULT, which is a LONG and cannot be filtered or fetched cheaply; only its
 430            // presence matters here.
 144431            command.CommandText =
 144432                """
 144433                SELECT COLUMN_NAME, DATA_TYPE, NULLABLE, CHAR_LENGTH, DATA_PRECISION, DATA_SCALE,
 144434                       DEFAULT_LENGTH, VIRTUAL_COLUMN, IDENTITY_COLUMN
 144435                FROM ALL_TAB_COLS
 144436                WHERE OWNER = :owner AND TABLE_NAME = :table_name AND HIDDEN_COLUMN = 'NO'
 144437                """;
 144438            command.Parameters.Add(new OracleParameter("owner", table.Owner));
 144439            command.Parameters.Add(new OracleParameter("table_name", table.Name));
 144440            await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 1153441            while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 442            {
 1009443                columns[reader.GetString(0)] = new ActualColumn(
 1009444                    DataType: reader.GetString(1),
 1009445                    Nullable: string.Equals(reader.GetString(2), "Y", StringComparison.OrdinalIgnoreCase),
 1009446                    CharLength: reader.IsDBNull(3) ? null : reader.GetInt64(3),
 1009447                    Precision: reader.IsDBNull(4) ? null : reader.GetInt64(4),
 1009448                    Scale: reader.IsDBNull(5) ? null : reader.GetInt64(5),
 1009449                    HasDefault: !reader.IsDBNull(6),
 1009450                    Virtual: !reader.IsDBNull(7) && string.Equals(reader.GetString(7), "YES", StringComparison.OrdinalIg
 1009451                    Identity: !reader.IsDBNull(8) && string.Equals(reader.GetString(8), "YES", StringComparison.OrdinalI
 452            }
 144453        }
 454
 2299455        foreach (var expected in ExpectedColumns)
 456        {
 1006457            if (!columns.TryGetValue(expected.Name, out var actual))
 458            {
 1459                throw new InvalidOperationException(
 1460                    $"The Oracle durable-flow table '{_options.TableName}' has no '{expected.Name}' column. It was creat
 1461                    "earlier build or by hand and does not match the shape this store reads and writes " +
 7462                    $"({string.Join(", ", ExpectedColumns.Select(column => $"{column.Name} {column.Declaration}"))}). Re
 1463                    "or add the missing columns — the DDL is in docs/durable-flow-state-stores.md.");
 464            }
 465
 1005466            if (expected.Mismatch(actual) is { } mismatch)
 467            {
 0468                throw new InvalidOperationException(
 0469                    $"The Oracle durable-flow table '{_options.TableName}' declares {expected.Name} as '{actual.DataType
 0470                    $"{(actual.Nullable ? " NULL" : " NOT NULL")}', which {mismatch}. This store needs " +
 0471                    $"{expected.Name} {expected.Declaration}. Fix it with " +
 0472                    $"ALTER TABLE {_options.TableName} MODIFY ({expected.Name} {expected.Declaration}); " +
 0473                    "(tables this build creates get that shape automatically).");
 474            }
 475        }
 476
 477        // Columns this store never names in an INSERT. One that the database cannot fill in for
 478        // itself makes EVERY create fail — the shape is otherwise perfect, so the failure arrives
 479        // at the first flow rather than at startup, which is the wrong end of the deployment.
 480        // Virtual and identity columns are fine; so is anything nullable or defaulted.
 2295481        foreach (var (name, actual) in columns)
 482        {
 5037483            if (ExpectedColumns.Any(expected => string.Equals(expected.Name, name, StringComparison.OrdinalIgnoreCase))
 1005484                || actual.IsWritableWithoutValue)
 485            {
 486                continue;
 487            }
 488
 1489            throw new InvalidOperationException(
 1490                $"The Oracle durable-flow table '{_options.TableName}' has an extra column '{name}' ({actual.DataType} N
 1491                "with no default. This store writes only its own columns, so every flow creation would fail on that colu
 1492                "it a default, make it nullable, virtual, or identity, or move it to a table of your own.");
 493        }
 494
 142495        await VerifyFlowIdIsUniqueAsync(connection, table.Owner, table.Name, cancellationToken).ConfigureAwait(false);
 139496        return true;
 140497    }
 498
 499    /// <summary>
 500    /// Resolves what the store's unqualified table name actually reaches — the same path Oracle's
 501    /// own name resolution takes: an object in CURRENT_SCHEMA, else a private synonym there, else
 502    /// a PUBLIC synonym, following synonym chains to the base object. USER_* views describe the
 503    /// CONNECTING user, which is the wrong scope whenever a logon trigger sets CURRENT_SCHEMA or
 504    /// the table is reached through a synonym: they either go blank (silently skipping every
 505    /// check) or report the synonym itself as a disqualifying non-TABLE. Returns null when nothing
 506    /// resolves (table absent) or the chain leaves this database (a DB link, unverifiable here);
 507    /// throws when the name resolves to a non-TABLE object.
 508    /// </summary>
 509    private async Task<(string Owner, string Name)?> ResolveBaseTableAsync(OracleConnection connection, CancellationToke
 510    {
 511        string? owner;
 146512        await using (var command = connection.CreateCommand())
 513        {
 146514            command.CommandText = "SELECT SYS_CONTEXT('USERENV', 'CURRENT_SCHEMA') FROM dual";
 146515            owner = (string?)await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 516        }
 517
 146518        if (string.IsNullOrEmpty(owner))
 0519            return null;
 520
 146521        var name = CatalogName;
 522        // Oracle itself raises ORA-01775 on a looping synonym chain; the bound only keeps a
 523        // broken catalog from spinning this check.
 294524        for (var hop = 0; hop < 10; hop++)
 525        {
 147526            var objectTypes = new List<string>();
 147527            await using (var command = connection.CreateCommand())
 528            {
 147529                command.BindByName = true;
 530                // Only the namespaces that can collide with a table: tables, views, materialized
 531                // views, synonyms, and sequences share one namespace (indexes and triggers do
 532                // not). A materialized view registers BOTH a TABLE and a MATERIALIZED VIEW row for
 533                // its name, so any non-TABLE row disqualifies it.
 147534                command.CommandText =
 147535                    """
 147536                    SELECT OBJECT_TYPE FROM ALL_OBJECTS
 147537                    WHERE OWNER = :owner AND OBJECT_NAME = :object_name
 147538                      AND OBJECT_TYPE IN ('TABLE', 'VIEW', 'MATERIALIZED VIEW', 'SYNONYM', 'SEQUENCE')
 147539                    """;
 147540                command.Parameters.Add(new OracleParameter("owner", owner));
 147541                command.Parameters.Add(new OracleParameter("object_name", name));
 147542                await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 293543                while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 146544                    objectTypes.Add(reader.GetString(0));
 147545            }
 546
 147547            if (objectTypes.Count == 0)
 548            {
 549                // Nothing in this schema: an unqualified name falls through to a PUBLIC synonym.
 1550                if (await ResolveSynonymAsync(connection, "PUBLIC", name, cancellationToken).ConfigureAwait(false) is no
 1551                    return null;
 552
 0553                (owner, name) = publicTarget;
 0554                continue;
 555            }
 556
 557            // Synonyms share the object namespace within a schema, so a SYNONYM row is the only
 558            // row: follow it.
 146559            if (objectTypes.Contains("SYNONYM"))
 560            {
 1561                if (await ResolveSynonymAsync(connection, owner, name, cancellationToken).ConfigureAwait(false) is not {
 0562                    return null;
 563
 1564                (owner, name) = target;
 1565                continue;
 566            }
 567
 290568            if (objectTypes.Find(type => !type.Equals("TABLE", StringComparison.OrdinalIgnoreCase)) is { } notATable)
 569            {
 1570                throw new InvalidOperationException(
 1571                    $"The Oracle durable-flow table name '{_options.TableName}' resolves to a {notATable} ({owner}.{name
 1572                    "it may even work, but this store's MERGE-based create needs a real table with a unique key on flow_
 1573                    "duplicate flows, so writes would fail — or double-run flows — mid-operation instead of at startup. 
 1574                    $"{nameof(OracleDurableFlowOptions)}.{nameof(OracleDurableFlowOptions.TableName)} at a table, or dro
 1575                    $"{notATable} and let the store create the table.");
 576            }
 577
 144578            return (owner, name);
 579        }
 580
 0581        throw new InvalidOperationException(
 0582            $"The Oracle durable-flow table name '{_options.TableName}' did not resolve to a base table within 10 synony
 0583            "the synonym chain is looping or degenerate.");
 145584    }
 585
 586    private static async Task<(string Owner, string Name)?> ResolveSynonymAsync(
 587        OracleConnection connection,
 588        string owner,
 589        string name,
 590        CancellationToken cancellationToken)
 591    {
 2592        await using var command = connection.CreateCommand();
 2593        command.BindByName = true;
 2594        command.CommandText =
 2595            """
 2596            SELECT TABLE_OWNER, TABLE_NAME, DB_LINK FROM ALL_SYNONYMS
 2597            WHERE OWNER = :owner AND SYNONYM_NAME = :synonym_name
 2598            """;
 2599        command.Parameters.Add(new OracleParameter("owner", owner));
 2600        command.Parameters.Add(new OracleParameter("synonym_name", name));
 2601        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 2602        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 1603            return null;
 604
 605        // A DB-link synonym points outside this database; the local catalog cannot describe it.
 1606        if (!await reader.IsDBNullAsync(2, cancellationToken).ConfigureAwait(false))
 0607            return null;
 608
 1609        if (await reader.IsDBNullAsync(0, cancellationToken).ConfigureAwait(false))
 0610            return null;
 611
 1612        return (reader.GetString(0), reader.GetString(1));
 2613    }
 614
 615    /// <summary>
 616    /// Requires ordinal NVARCHAR2 comparison in this session. NLS_COMP=BINARY (Oracle's default)
 617    /// compares bytes regardless of NLS_SORT; NLS_COMP=LINGUISTIC (or the deprecated ANSI) routes
 618    /// every comparison through NLS_SORT instead, where anything but BINARY — BINARY_CI,
 619    /// BINARY_AI, a language sort — folds case or accents. Sessions inherit these from instance
 620    /// parameters, client NLS configuration, and logon triggers, uniformly for every connection
 621    /// this store opens, so one check on one pooled session stands for all of them. Internal (not
 622    /// private) so the integration suite can run it against a deliberately mis-set session without
 623    /// installing a logon trigger.
 624    /// </summary>
 625    internal static async Task VerifyComparisonSemanticsAsync(OracleConnection connection, CancellationToken cancellatio
 626    {
 148627        string? comp = null;
 148628        string? sort = null;
 148629        await using (var command = connection.CreateCommand())
 630        {
 148631            command.CommandText = "SELECT PARAMETER, VALUE FROM NLS_SESSION_PARAMETERS WHERE PARAMETER IN ('NLS_COMP', '
 148632            await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 444633            while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 634            {
 296635                if (string.Equals(reader.GetString(0), "NLS_COMP", StringComparison.OrdinalIgnoreCase))
 148636                    comp = reader.GetString(1);
 637                else
 148638                    sort = reader.GetString(1);
 639            }
 148640        }
 641
 148642        if (DiagnoseComparisonSemantics(comp, sort) is { } linguisticSession)
 1643            throw new InvalidOperationException(linguisticSession);
 147644    }
 645
 646    /// <summary>
 647    /// The comparison-semantics decision on its own catalog inputs: the rejection message for a
 648    /// session whose NLS settings fold flow-id equality, or null for an ordinal one. Folded
 649    /// equality is the silent kind of wrong — every <c>WHERE flow_id = :flow_id</c> this store
 650    /// runs matches BOTH casings, so a load returns another flow's state and a lease update fences
 651    /// the other flow's execution, while the primary-key index (always binary) keeps admitting
 652    /// both rows. Nothing errors; flows cross-route. A session whose parameters could not be read
 653    /// is not known to be ordinal, so silence does not pass.
 654    /// </summary>
 655    internal static string? DiagnoseComparisonSemantics(string? nlsComp, string? nlsSort)
 656    {
 179657        static bool IsBinary(string? value) => string.Equals(value, "BINARY", StringComparison.OrdinalIgnoreCase);
 166658        if (IsBinary(nlsComp) || IsBinary(nlsSort))
 155659            return null;
 660
 11661        return
 11662            $"This Oracle session compares text linguistically (NLS_COMP='{nlsComp ?? "(unknown)"}', " +
 11663            $"NLS_SORT='{nlsSort ?? "(unknown)"}'), so flow ids differing only in case (or accent) match each " +
 11664            "other's rows: a load can return the other flow's state and a lease can fence the other flow's " +
 11665            "execution, while the binary primary-key index still admits both ids. Flow ids are compared " +
 11666            "ordinally by the engine. Restore binary comparison with ALTER SESSION SET NLS_COMP = BINARY (or " +
 11667            "NLS_SORT = BINARY) — typically by removing the logon trigger or client NLS configuration that " +
 11668            "changed them.";
 669    }
 670
 671    /// <summary>One column as <c>USER_TAB_COLS</c> reports it. Internal (with the expected shape
 672    /// below) so the accept/reject decision table is unit-testable without an Oracle server — the
 673    /// integration suite proves the same decisions against a real catalog, but only where the
 674    /// Oracle container runs.</summary>
 675    internal readonly record struct ActualColumn(
 1507676        string DataType,
 1051677        bool Nullable,
 295678        long? CharLength,
 151679        long? Precision,
 441680        long? Scale,
 11681        bool HasDefault,
 7682        bool Virtual,
 5683        bool Identity)
 684    {
 685        /// <summary>
 686        /// Whether this store could insert a row without naming this column. True when the column
 687        /// is nullable, carries a default, is computed by the database, or is an identity.
 688        /// </summary>
 14689        internal bool IsWritableWithoutValue => Nullable || HasDefault || Virtual || Identity;
 690    }
 691
 692    /// <summary>
 693    /// What this store needs from one column, and the check that says so. Widths and precisions
 694    /// are MINIMA rather than exact matches: a wider flow_id or a higher-precision timestamp still
 695    /// satisfies every promise the store makes, and rejecting a more generous schema would be a
 696    /// false alarm. Too NARROW is not — NVARCHAR2(10) passes a name-only check and then errors on
 697    /// the first 400-character id the public contract permits.
 698    /// </summary>
 10029699    internal sealed record ExpectedColumn(string Name, string Declaration, string DataType, bool Nullable, long? Minimum
 700    {
 701        internal string? Mismatch(ActualColumn actual)
 702        {
 1049703            if (string.Equals(DataType, "TIMESTAMP", StringComparison.Ordinal))
 704            {
 705                // USER_TAB_COLS embeds the fractional precision in the type name ('TIMESTAMP(6)'),
 706                // so the family is a prefix match. The WITH [LOCAL] TIME ZONE variants are
 707                // rejected as different types: their values shift with the session time zone, and
 708                // expiry/lease math must stay on the plain UTC timestamps this store writes.
 447709                if (!actual.DataType.StartsWith("TIMESTAMP", StringComparison.OrdinalIgnoreCase)
 447710                    || actual.DataType.Contains("TIME ZONE", StringComparison.OrdinalIgnoreCase))
 711                {
 6712                    return $"is a '{actual.DataType}'";
 713                }
 714            }
 602715            else if (!string.Equals(actual.DataType, DataType, StringComparison.OrdinalIgnoreCase))
 716            {
 6717                return $"is a '{actual.DataType}'";
 718            }
 719
 1037720            if (actual.Nullable != Nullable)
 4721                return Nullable ? "is NOT NULL (this store writes NULL to it)" : "is nullable";
 1033722            if (Minimum is not { } minimum)
 146723                return null;
 724
 725            // CHAR_LENGTH for the character columns, DATA_SCALE (fractional-second digits) for the
 726            // timestamps, DATA_PRECISION for NUMBER: a TIMESTAMP(0) column silently rounds the
 727            // sub-second lease arithmetic this store runs on SYS_EXTRACT_UTC(SYSTIMESTAMP), which
 728            // is how two workers end up holding one lease, and a NUMBER(9) revision overflows
 729            // without a word. An unconstrained NUMBER (null precision) holds 38 digits and passes.
 887730            var actualSize = DataType switch
 887731            {
 441732                "TIMESTAMP" => actual.Scale,
 151733                "NUMBER" => actual.Precision ?? 38,
 295734                _ => actual.CharLength
 887735            };
 887736            return actualSize is { } size && size >= minimum
 887737                ? null
 887738                : $"holds {actualSize?.ToString() ?? "an unknown size"} where at least {minimum} is required";
 739        }
 740    }
 741
 742    /// <summary>
 743    /// The shape this store reads and writes. flow_id, lease_id, and state_json are the NATIONAL
 744    /// character types on purpose: NVARCHAR2/NCLOB store the national character set, which is
 745    /// always Unicode, while VARCHAR2/CLOB inherit the database character set — on a non-AL32UTF8
 746    /// database a perfectly legal flow id (an emoji, most non-Latin text) mangles or fails on
 747    /// insert, so those types are rejected rather than trusted. No default expressions are
 748    /// verified because none are load-bearing: every write names every column except the two lease
 749    /// fields, whose absence means NULL.
 750    /// </summary>
 6751    internal static readonly ExpectedColumn[] ExpectedColumns =
 6752    [
 6753        new("flow_id", "NVARCHAR2(400) NOT NULL", "NVARCHAR2", Nullable: false, Minimum: 400),
 6754        new("state_json", "NCLOB NOT NULL", "NCLOB", Nullable: false),
 6755        new("expires_at_utc", "TIMESTAMP(6) NOT NULL", "TIMESTAMP", Nullable: false, Minimum: 6),
 6756        new("updated_at_utc", "TIMESTAMP(6) NOT NULL", "TIMESTAMP", Nullable: false, Minimum: 6),
 6757        new("revision", "NUMBER(19) DEFAULT 0 NOT NULL", "NUMBER", Nullable: false, Minimum: 19),
 6758        new("lease_id", "NVARCHAR2(64) NULL", "NVARCHAR2", Nullable: true, Minimum: 64),
 6759        new("lease_expires_at_utc", "TIMESTAMP(6) NULL", "TIMESTAMP", Nullable: true, Minimum: 6)
 6760    ];
 761
 762    /// <summary>
 763    /// Requires a unique key on the WHOLE of flow_id and nothing else. The PRIMARY KEY this
 764    /// store's DDL declares is the usual shape, but any enabled single-column UNIQUE constraint —
 765    /// or a bare unique INDEX, which raises ORA-00001 without a constraint row — serves the
 766    /// MERGE's duplicate detection, so all of them are accepted. A COMPOSITE key is not: it
 767    /// permits two rows with one flow_id. A DISABLED constraint is not either: it sits in the
 768    /// catalog and enforces nothing. (Deferrable constraints use a NONUNIQUE index, which is why
 769    /// the constraint and index arms are both asked.)
 770    /// </summary>
 771    private async Task VerifyFlowIdIsUniqueAsync(OracleConnection connection, string owner, string tableName, Cancellati
 772    {
 142773        await using var command = connection.CreateCommand();
 142774        command.BindByName = true;
 142775        command.CommandText =
 142776            """
 142777            SELECT 1 FROM dual
 142778            WHERE EXISTS (
 142779                SELECT 1 FROM ALL_CONSTRAINTS c
 142780                WHERE c.OWNER = :owner AND c.TABLE_NAME = :table_name AND c.CONSTRAINT_TYPE IN ('P', 'U') AND c.STATUS =
 142781                  AND EXISTS (SELECT 1 FROM ALL_CONS_COLUMNS k
 142782                              WHERE k.OWNER = c.OWNER AND k.CONSTRAINT_NAME = c.CONSTRAINT_NAME AND k.COLUMN_NAME = 'FLO
 142783                  AND NOT EXISTS (SELECT 1 FROM ALL_CONS_COLUMNS o
 142784                                  WHERE o.OWNER = c.OWNER AND o.CONSTRAINT_NAME = c.CONSTRAINT_NAME AND o.COLUMN_NAME <>
 142785               OR EXISTS (
 142786                SELECT 1 FROM ALL_INDEXES i
 142787                WHERE i.TABLE_OWNER = :owner AND i.TABLE_NAME = :table_name AND i.UNIQUENESS = 'UNIQUE'
 142788                  AND EXISTS (SELECT 1 FROM ALL_IND_COLUMNS k
 142789                              WHERE k.INDEX_OWNER = i.OWNER AND k.INDEX_NAME = i.INDEX_NAME AND k.COLUMN_NAME = 'FLOW_ID
 142790                  AND NOT EXISTS (SELECT 1 FROM ALL_IND_COLUMNS o
 142791                                  WHERE o.INDEX_OWNER = i.OWNER AND o.INDEX_NAME = i.INDEX_NAME AND o.COLUMN_NAME <> 'FL
 142792            """;
 142793        command.Parameters.Add(new OracleParameter("owner", owner));
 142794        command.Parameters.Add(new OracleParameter("table_name", tableName));
 795
 142796        if (await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false) is not null)
 797            return;
 798
 3799        throw new InvalidOperationException(
 3800            $"The Oracle durable-flow table '{_options.TableName}' has no enabled unique key on the whole of flow_id. St
 3801            "flow is an insert-if-absent, and this store's MERGE learns that a ledger already exists from the duplicate-
 3802            "ORA-00001 when two starts race. Without such a key nothing reports the duplicate, so two concurrent starts 
 3803            $"flow id both insert and the flow runs twice. Fix it with ALTER TABLE {_options.TableName} ADD PRIMARY KEY 
 3804            "(tables this build creates declare it automatically).");
 139805    }
 806
 807    private async Task<bool> UpdateLeaseAsync(
 808        string flowId,
 809        string leaseId,
 810        TimeSpan leaseDuration,
 811        bool acquire,
 812        CancellationToken cancellationToken)
 813    {
 165814        DurableFlowStoreShared.ValidateLeaseArgs(flowId, leaseId, leaseDuration);
 815
 163816        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 163817        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 163818        await using var command = connection.CreateCommand();
 163819        command.BindByName = true;
 820        // Lease fencing runs entirely on the database clock: acquire steals only leases the
 821        // database considers expired, and renew/extend stays relative to the server's UTC time,
 822        // so worker clock skew can never make two nodes hold the same lease.
 163823        command.CommandText =
 163824            $"""
 163825            UPDATE {Table}
 163826            SET lease_id = :lease_id, lease_expires_at_utc = {AddMilliseconds(":lease_ms")}
 163827            WHERE flow_id = :flow_id
 163828              AND expires_at_utc > {UtcNowSql}
 163829              AND {(acquire ? $"(lease_id IS NULL OR lease_expires_at_utc <= {UtcNowSql} OR lease_id = :lease_id)" : $"l
 163830            """;
 163831        command.Parameters.Add(NationalId("flow_id", flowId));
 163832        command.Parameters.Add(NationalId("lease_id", leaseId));
 163833        command.Parameters.Add(new OracleParameter("lease_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(leaseDu
 163834        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 163835    }
 836
 837    private Task<OracleConnection> OpenConnectionAsync(CancellationToken cancellationToken)
 2470838        => DurableFlowStoreShared.OpenConnectionAsync<OracleConnection>(_options.ConnectionString, cancellationToken);
 839
 2636840    private string Table => _options.TableName;
 137841    private string IndexName => DurableFlowStoreShared.DerivedName(_options.TableName, "_EXPIRES_IDX", 128);
 842
 843    /// <summary>
 844    /// The name the data dictionary stores. This store interpolates <see cref="Table"/> unquoted,
 845    /// which Oracle resolves to the upper-cased catalog entry; the validated identifier alphabet
 846    /// (ASCII letters, digits, underscore) makes the invariant upper-casing exact, so catalog
 847    /// lookups on this name describe precisely the object the store's SQL touches.
 848    /// </summary>
 146849    private string CatalogName => _options.TableName.ToUpperInvariant();
 850}
 851}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

AddMilliseconds(System.String)
.ctor(Microsoft.Extensions.Options.IOptions`1<AsyncResponse.DurableFlows.Oracle.OracleDurableFlowOptions>,Microsoft.Extensions.Logging.ILogger`1<AsyncResponse.DurableFlows.Oracle.OracleFlowStateStore>)
LoadAsync()
ValidateCreate(System.String,AsyncResponse.FlowState,System.TimeSpan)
TryCreateAsync()
TryCreateCoreAsync()
TryUpdateAsync()
TryAcquireLeaseAsync(System.String,System.String,System.TimeSpan,System.Threading.CancellationToken)
TryRenewLeaseAsync(System.String,System.String,System.TimeSpan,System.Threading.CancellationToken)
ReleaseLeaseAsync()
ObserveLeaseAsync()
TryDeleteAsync()
PruneExpiredAsync()
EnsureCreatedAsync()
NationalId(System.String,System.String)
ExecuteIgnoringExistsAsync()
VerifyFlowTableAsync()
ResolveBaseTableAsync()
ResolveSynonymAsync()
VerifyComparisonSemanticsAsync()
IsBinary()
DiagnoseComparisonSemantics(System.String,System.String)
get_DataType()
get_Nullable()
get_CharLength()
get_Precision()
get_Scale()
get_HasDefault()
get_Virtual()
get_Identity()
get_IsWritableWithoutValue()
get_Name()
Mismatch(AsyncResponse.DurableFlows.Oracle.OracleFlowStateStore/ActualColumn)
.cctor()
VerifyFlowIdIsUniqueAsync()
UpdateLeaseAsync()
OpenConnectionAsync(System.Threading.CancellationToken)
get_Table()
get_IndexName()
get_CatalogName()