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

Information
Class: AsyncResponse.DurableFlows.PostgreSQL.PostgreSqlDurableFlowOptions
Assembly: AsyncResponse.DurableFlows.PostgreSQL
File(s): /_/src/DurableFlows/AsyncResponse.DurableFlows.PostgreSQL/PostgreSqlDurableFlows.cs
Line coverage
100%
Covered lines: 15
Uncovered lines: 0
Coverable lines: 15
Total lines: 497
Line coverage: 100%
Branch coverage
100%
Covered branches: 2
Total branches: 2
Branch coverage: 100%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
get_ConnectionString()100%11100%
get_SchemaName()100%11100%
get_TableName()100%11100%
get_AutoCreateSchema()100%11100%
get_PruneInterval()100%11100%
get_PruneBudget()100%11100%
get_MaxStateBytes()100%11100%
Validate()100%22100%

File(s)

/_/src/DurableFlows/AsyncResponse.DurableFlows.PostgreSQL/PostgreSqlDurableFlows.cs

#LineLine coverage
 1using AsyncResponse;
 2using AsyncResponse.DurableFlows.Internal;
 3using AsyncResponse.DurableFlows.PostgreSQL;
 4using Microsoft.Extensions.DependencyInjection;
 5using Microsoft.Extensions.DependencyInjection.Extensions;
 6using Microsoft.Extensions.Logging;
 7using Microsoft.Extensions.Options;
 8using Npgsql;
 9using NpgsqlTypes;
 10
 11namespace Microsoft.Extensions.DependencyInjection
 12{
 13    /// <summary>DI registration for the PostgreSQL durable-flow state store.</summary>
 14    public static class PostgreSqlDurableFlowServiceCollectionExtensions
 15    {
 16        /// <summary>
 17        /// Stores durable-flow state in PostgreSQL. Hosts may either register an
 18        /// <see cref="NpgsqlDataSource"/> singleton or set
 19        /// <see cref="PostgreSqlDurableFlowOptions.ConnectionString"/>.
 20        /// </summary>
 21        public static AsyncResponseRegistrationBuilder WithPostgreSqlDurableFlows(
 22            this AsyncResponseRegistrationBuilder builder,
 23            Action<PostgreSqlDurableFlowOptions>? configure = null)
 24        {
 25            // Singleton on purpose: schema provisioning is cached per store instance, and the
 26            // executor resolves the store from a fresh scope per flow execution — a scoped store
 27            // would re-run EnsureCreated's DDL round-trip on every run. All dependencies are
 28            // singletons, so the singleton is safe.
 29            builder.Services.TryAddSingleton(provider =>
 30            {
 31                var options = provider.GetRequiredService<IOptions<PostgreSqlDurableFlowOptions>>();
 32
 33                // Reuse a host-registered NpgsqlDataSource when present; otherwise create one from
 34                // ConnectionString, owned (and disposed) by the store. Nothing is registered as a
 35                // bare NpgsqlDataSource service, so unrelated resolutions of that type are never
 36                // answered — or broken — by this package.
 37                var shared = provider.GetService<NpgsqlDataSource>();
 38                var logger = provider.GetService<ILogger<PostgreSqlFlowStateStore>>();
 39                if (shared is not null)
 40                    return new PostgreSqlFlowStateStore(shared, options, logger: logger);
 41
 42                if (string.IsNullOrWhiteSpace(options.Value.ConnectionString))
 43                    throw new InvalidOperationException($"{nameof(PostgreSqlDurableFlowOptions)}.{nameof(PostgreSqlDurab
 44                return new PostgreSqlFlowStateStore(NpgsqlDataSource.Create(options.Value.ConnectionString), options, ow
 45            });
 46            return builder.WithDurableFlows<PostgreSqlFlowStateStore, PostgreSqlDurableFlowOptions>(configure);
 47        }
 48    }
 49}
 50
 51namespace AsyncResponse.DurableFlows.PostgreSQL
 52{
 53/// <summary>Options for the PostgreSQL durable-flow state store.</summary>
 54public sealed class PostgreSqlDurableFlowOptions : DurableFlowOptions
 55{
 56    /// <summary>Optional PostgreSQL connection string used when no <see cref="NpgsqlDataSource"/> is registered.</summa
 1057    public string? ConnectionString { get; set; }
 58
 59    /// <summary>Database schema that contains the durable-flow table. Default: <c>public</c>.</summary>
 437660    public string SchemaName { get; set; } = "public";
 61
 62    /// <summary>Table storing one durable-flow ledger row per flow id.</summary>
 469563    public string TableName { get; set; } = "asyncresponse_flow_state";
 64
 65    /// <summary>Creates the schema, table, and expiry index on first use.</summary>
 36966    public bool AutoCreateSchema { get; set; } = true;
 67
 68    /// <summary>
 69    /// How often <see cref="PostgreSqlFlowStateStore.TryCreateAsync"/> opportunistically deletes one
 70    /// bounded batch (1000 rows) of expired rows (loads already treat expired state as absent;
 71    /// pruning bounds table growth). Zero or negative prunes on every save. Default: 5 minutes.
 72    /// </summary>
 51473    public TimeSpan PruneInterval { get; set; } = TimeSpan.FromMinutes(5);
 74
 75    /// <summary>
 76    /// Wall-clock budget one opportunistic prune may spend draining expired rows in batches of
 77    /// 1000 after its first batch (the first always runs). A single batch per interval capped
 78    /// cleanup at ~3 rows/second, which any busier instance outgrew forever; the prune now drains
 79    /// batches until one comes back short or this budget lapses, and reports the outcome on the
 80    /// <c>AsyncResponse</c> meter (<c>asyncresponse.flow_state.pruned_rows</c>,
 81    /// <c>prune_failures</c>, <c>prune_budget_exhausted</c>) and the store's logger. The create
 82    /// that triggers the prune waits for it, so this bounds that create's added latency. Zero
 83    /// keeps the historical single batch. Default: 2 seconds.
 84    /// </summary>
 58485    public TimeSpan PruneBudget { get; set; } = DurableFlowStoreShared.DefaultPruneBudget;
 86
 87    /// <summary>
 88    /// Maximum serialized flow-state size in bytes accepted by writes; oversized ledgers fail fast
 89    /// with an actionable error instead of an opaque provider error. Default: <c>null</c>
 90    /// (unlimited — PostgreSQL <c>jsonb</c> is effectively unbounded), settable as an operator budget.
 91    /// </summary>
 152192    public long? MaxStateBytes { get; set; }
 93
 94    /// <summary>Validates option values and throws on misconfiguration.</summary>
 95    public void Validate()
 96    {
 22597        DurableFlowStoreShared.ValidateIdentifier(SchemaName, $"{nameof(PostgreSqlDurableFlowOptions)}.{nameof(SchemaNam
 22598        DurableFlowStoreShared.ValidateIdentifier(TableName, $"{nameof(PostgreSqlDurableFlowOptions)}.{nameof(TableName)
 99
 100        // Indexes share PostgreSQL's relation namespace with tables: a table whose name ends
 101        // exactly where the reserved "_expires_idx" stem truncates derives its own name, and
 102        // CREATE INDEX IF NOT EXISTS would silently match the table and skip the index.
 223103        if (string.Equals(DurableFlowStoreShared.DerivedName(TableName, "_expires_idx", 63), TableName, StringComparison
 2104            throw new InvalidOperationException(
 2105                $"{nameof(PostgreSqlDurableFlowOptions)}.{nameof(TableName)} '{TableName}' collides with its derived exp
 221106        DurableFlowStoreShared.ValidateMaxStateBytes(MaxStateBytes, nameof(PostgreSqlDurableFlowOptions));
 219107        DurableFlowStoreShared.ValidatePruneBudget(PruneBudget, nameof(PostgreSqlDurableFlowOptions));
 217108    }
 109}
 110
 111/// <summary>PostgreSQL implementation of <see cref="IFlowStateStore"/>.</summary>
 112public sealed class PostgreSqlFlowStateStore : IFlowStateStore, IDisposable, IAsyncDisposable
 113{
 114    private readonly ILogger<PostgreSqlFlowStateStore>? _logger;
 115
 116    private readonly NpgsqlDataSource _dataSource;
 117    private readonly PostgreSqlDurableFlowOptions _options;
 118    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 119    private readonly bool _ownsDataSource;
 120    private readonly long _schemaLockKey;
 121    private long _lastPruneTicks;
 122    private volatile bool _created;
 123
 124    public PostgreSqlFlowStateStore(NpgsqlDataSource dataSource, IOptions<PostgreSqlDurableFlowOptions> options, bool ow
 125    {
 126        _dataSource = dataSource;
 127        _logger = logger;
 128        _options = options.Value;
 129        _options.Validate();
 130        _ownsDataSource = ownsDataSource;
 131        _schemaLockKey = DurableFlowStoreShared.SchemaLockKey(_options.SchemaName);
 132    }
 133
 134    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 135    {
 136        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 137        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 138
 139        // All expiry/lease time math in this store runs on the database clock (now()), never an
 140        // app-computed timestamp: with multiple workers, app clock skew beyond the lease window
 141        // would let two nodes both consider a lease expired and double-run a flow.
 142        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 143        await using var command = connection.CreateCommand();
 144        command.CommandText = $"SELECT state_json::text, revision FROM {Table} WHERE flow_id = @flow_id AND expires_at_u
 145        command.Parameters.AddWithValue("flow_id", flowId);
 146
 147        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 148        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 149            return null;
 150
 151        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 152    }
 153
 154    /// <inheritdoc />
 155    public void ValidateCreate(string flowId, FlowState state, TimeSpan ttl)
 156    {
 157        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 158        if (_options.MaxStateBytes is not null)
 159            _ = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "PostgreSQL");
 160    }
 161
 162    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 163    {
 164        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 165        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "PostgreSQL");
 166        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 167        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 168            await DurableFlowStoreShared.PruneQuietlyAsync(() => PruneExpiredAsync(cancellationToken), _options.PruneBud
 169
 170        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 171        await using var command = connection.CreateCommand();
 172        command.CommandText =
 173            $"""
 174            INSERT INTO {Table} (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 175            VALUES (@flow_id, @state_json, now() + @ttl, now(), @revision)
 176            ON CONFLICT (flow_id) DO UPDATE
 177            SET state_json = EXCLUDED.state_json,
 178                expires_at_utc = EXCLUDED.expires_at_utc,
 179                updated_at_utc = EXCLUDED.updated_at_utc,
 180                revision = EXCLUDED.revision,
 181                lease_id = NULL,
 182                lease_expires_at_utc = NULL
 183            WHERE {Table}.expires_at_utc <= now();
 184            """;
 185        command.Parameters.AddWithValue("flow_id", flowId);
 186        command.Parameters.Add("state_json", NpgsqlDbType.Text).Value = stateJson;
 187        command.Parameters.Add("ttl", NpgsqlDbType.Interval).Value = DurableFlowStoreShared.ServerClockTtl(ttl);
 188        command.Parameters.AddWithValue("revision", state.Revision);
 189        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 190    }
 191
 192    public async Task<bool> TryUpdateAsync(
 193        string flowId,
 194        FlowState state,
 195        long expectedRevision,
 196        TimeSpan ttl,
 197        string? leaseId = null,
 198        CancellationToken cancellationToken = default)
 199    {
 200        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 201        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "PostgreSQL");
 202        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 203
 204        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 205        await using var command = connection.CreateCommand();
 206        command.CommandText =
 207            $"""
 208            UPDATE {Table}
 209            SET state_json = @state_json,
 210                expires_at_utc = now() + @ttl,
 211                updated_at_utc = now(),
 212                revision = @new_revision
 213            WHERE flow_id = @flow_id
 214              AND revision = @expected_revision
 215              AND expires_at_utc > now()
 216              AND (@lease_id IS NULL OR (lease_id = @lease_id AND lease_expires_at_utc > now()));
 217            """;
 218        command.Parameters.AddWithValue("flow_id", flowId);
 219        command.Parameters.Add("state_json", NpgsqlDbType.Text).Value = stateJson;
 220        command.Parameters.Add("ttl", NpgsqlDbType.Interval).Value = DurableFlowStoreShared.ServerClockTtl(ttl);
 221        command.Parameters.AddWithValue("expected_revision", expectedRevision);
 222        command.Parameters.AddWithValue("new_revision", state.Revision);
 223        command.Parameters.AddWithValue("lease_id", NpgsqlDbType.Text, (object?)leaseId ?? DBNull.Value);
 224        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 225    }
 226
 227    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 228        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 229
 230    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 231        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 232
 233    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 234    {
 235        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 236        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 237        await using var command = connection.CreateCommand();
 238        command.CommandText = $"UPDATE {Table} SET lease_id = NULL, lease_expires_at_utc = NULL WHERE flow_id = @flow_id
 239        command.Parameters.AddWithValue("flow_id", flowId);
 240        command.Parameters.AddWithValue("lease_id", leaseId);
 241        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 242    }
 243
 244    /// <inheritdoc />
 245    public async Task<FlowLeaseObservation?> ObserveLeaseAsync(string flowId, CancellationToken cancellationToken = defa
 246    {
 247        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 248        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 249
 250        // The two lease columns exactly as stored — deliberately no now() predicate, unlike every
 251        // other statement in this store: an expired lease nobody has taken over must keep reading
 252        // as the same lease, because the engine's proof of a live holder is that two observations
 253        // DIFFER. Whether it has lapsed stays UpdateLeaseAsync's call, on the database clock.
 254        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 255        await using var command = connection.CreateCommand();
 256        command.CommandText = $"SELECT lease_id, lease_expires_at_utc FROM {Table} WHERE flow_id = @flow_id;";
 257        command.Parameters.AddWithValue("flow_id", flowId);
 258
 259        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 260        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 261            return FlowLeaseObservation.Unheld;
 262
 263        return DurableFlowStoreShared.LeaseObservation(
 264            reader.IsDBNull(0) ? null : reader.GetString(0),
 265            reader.IsDBNull(1) ? null : reader.GetDateTime(1));
 266    }
 267
 268    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 269    {
 270        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 271        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 272
 273        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 274        await using var command = connection.CreateCommand();
 275        command.CommandText = $"DELETE FROM {Table} WHERE flow_id = @flow_id;";
 276        command.Parameters.AddWithValue("flow_id", flowId);
 277        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 278    }
 279
 280    private async Task<int> PruneExpiredAsync(CancellationToken cancellationToken)
 281    {
 282        // One bounded batch per call; DurableFlowStoreShared.PruneQuietlyAsync repeats it under
 283        // the PruneBudget while batches come back full (policy shared by all relational stores): an
 284        // unbatched DELETE over a large expired backlog holds row locks and bloats one
 285        // transaction for the unlucky create that triggered the prune. Loads already filter on
 286        // expiry, so any backlog beyond the batch just waits for the next interval.
 287        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 288        await using var command = connection.CreateCommand();
 289        command.CommandText =
 290            $"""
 291            DELETE FROM {Table}
 292            WHERE ctid IN (SELECT ctid FROM {Table} WHERE expires_at_utc <= now() LIMIT {DurableFlowStoreShared.PruneBat
 293            """;
 294        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 295    }
 296
 297    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 298    {
 299        if (_created)
 300            return;
 301
 302        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 303        try
 304        {
 305            if (_created)
 306                return;
 307
 308            await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 309            await using var transaction = await connection.BeginTransactionAsync(cancellationToken).ConfigureAwait(false
 310
 311            if (_options.AutoCreateSchema)
 312            {
 313                // Serialize schema creation across processes. CREATE ... IF NOT EXISTS is not atomic
 314                // against a concurrent create of the same object: two instances starting together both
 315                // pass the existence check and collide on the system catalog ("duplicate key ...
 316                // pg_type_typname_nsp_index"). The transaction-scoped advisory lock (keyed by schema,
 317                // shared with the channel/transport packages) lets one instance build the schema while
 318                // the rest wait and then find it already present.
 319                await using (var lockCommand = connection.CreateCommand())
 320                {
 321                    lockCommand.Transaction = transaction;
 322                    lockCommand.CommandText = "SELECT pg_advisory_xact_lock(@lock_key);";
 323                    lockCommand.Parameters.AddWithValue("lock_key", _schemaLockKey);
 324                    await lockCommand.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 325                }
 326
 327                await using var command = connection.CreateCommand();
 328                command.Transaction = transaction;
 329                command.CommandText =
 330                    $"""
 331                    CREATE SCHEMA IF NOT EXISTS {Schema};
 332                    CREATE TABLE IF NOT EXISTS {Table} (
 333                        flow_id text NOT NULL PRIMARY KEY,
 334                        state_json text NOT NULL,
 335                        expires_at_utc timestamptz NOT NULL,
 336                        updated_at_utc timestamptz NOT NULL,
 337                        revision bigint NOT NULL DEFAULT 0,
 338                        lease_id text NULL,
 339                        lease_expires_at_utc timestamptz NULL
 340                    );
 341                    CREATE INDEX IF NOT EXISTS {IndexName} ON {Table} (expires_at_utc);
 342
 343                    -- jsonb REJECTS the \u0000 escape System.Text.Json emits for U+0000 (SQLSTATE
 344                    -- 22P05), so a ledger every other store accepts failed every write here: the
 345                    -- flow could not start, or its checkpoint failed on every attempt until the
 346                    -- job dead-lettered. Nothing queries INSIDE the ledger (it is read back with
 347                    -- ::text), so text costs nothing. Guarded so the rewrite happens once.
 348                    DO $$
 349                    BEGIN
 350                        IF EXISTS (
 351                            SELECT 1 FROM information_schema.columns
 352                            WHERE table_schema = {SqlLiteral(_options.SchemaName)} AND table_name = {SqlLiteral(_options
 353                              AND column_name = 'state_json' AND data_type = 'jsonb')
 354                        THEN
 355                            ALTER TABLE {Table} ALTER COLUMN state_json TYPE text USING state_json::text;
 356                        END IF;
 357                    END $$;
 358                    """;
 359                try
 360                {
 361                    await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 362                }
 363                catch (PostgresException ex) when (ex.SqlState is PostgresErrorCodes.WrongObjectType or PostgresErrorCod
 364                {
 365                    // E.g. CREATE INDEX ... ON a name that is really another component's index:
 366                    // IF NOT EXISTS skipped the table create, and the dependent statement then hits
 367                    // the wrong relation kind mid-batch — surface the namespace collision instead of
 368                    // the raw "cannot open relation".
 369                    throw new InvalidOperationException(AsyncResponse.Internal.PostgreSqlRelationVerifier.DdlCollisionMe
 370                }
 371            }
 372            else if (!await TableExistsAsync(connection, transaction, cancellationToken).ConfigureAwait(false))
 373            {
 374                // Operator-managed schema and the migration has not run yet: the first query
 375                // surfaces a clear PostgreSQL error (the documented "create it yourself, later"
 376                // workflow), and _created stays unlatched so a later operation re-verifies once
 377                // the migration lands. When the relation DOES exist it flows into the same catalog
 378                // verification the DDL path uses — operator-provisioned schemas are exactly what
 379                // that check exists for.
 380                return;
 381            }
 382
 383            // The flow store can share a schema with the channel and transport stores (and
 384            // unrelated objects), whose derived names its own validation cannot see — and
 385            // IF NOT EXISTS also accepts a same-name index with the WRONG definition, exactly as
 386            // an operator-provisioned table can carry the wrong shape. Verify against the catalog
 387            // that both relations actually ARE what this store reads and writes, definitions
 388            // included (under the shared DDL lock when this build just ran the DDL).
 389            await AsyncResponse.Internal.PostgreSqlRelationVerifier.VerifyAsync(
 390                connection,
 391                transaction,
 392                _options.SchemaName,
 393                "durable-flow",
 394                [
 395                    new(_options.TableName, 'r', Columns:
 396                        [
 397                            new("flow_id", "text", Nullable: false, RequiresDeterministicCollation: true),
 398                            new("state_json", "text", Nullable: false),
 399                            new("expires_at_utc", "timestamp with time zone", Nullable: false),
 400                            new("updated_at_utc", "timestamp with time zone", Nullable: false),
 401                            new("revision", "bigint", Nullable: false, DefaultExpression: "0"),
 402                            new("lease_id", "text", Nullable: true),
 403                            new("lease_expires_at_utc", "timestamp with time zone", Nullable: true),
 404                        ], PrimaryKey: ["flow_id"]),
 405                    new(DurableFlowStoreShared.DerivedName(_options.TableName, "_expires_idx", 63), 'i', _options.TableN
 406                ],
 407                cancellationToken).ConfigureAwait(false);
 408
 409            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 410            _created = true;
 411        }
 412        finally
 413        {
 414            _ensureGate.Release();
 415        }
 416    }
 417
 418    /// <summary>
 419    /// Reports whether ANY relation occupies the configured name (any relkind: a view or foreign
 420    /// component's object must reach verification, which names the precise wrong-kind reason
 421    /// instead of skipping the checks).
 422    /// </summary>
 423    private async Task<bool> TableExistsAsync(NpgsqlConnection connection, NpgsqlTransaction transaction, CancellationTo
 424    {
 425        await using var command = connection.CreateCommand();
 426        command.Transaction = transaction;
 427        command.CommandText =
 428            """
 429            SELECT EXISTS (
 430                SELECT 1
 431                FROM pg_catalog.pg_class c
 432                JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
 433                WHERE n.nspname = @schema AND c.relname = @table);
 434            """;
 435        command.Parameters.AddWithValue("schema", _options.SchemaName);
 436        command.Parameters.AddWithValue("table", _options.TableName);
 437        return (bool)(await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false))!;
 438    }
 439
 440
 441    private async Task<bool> UpdateLeaseAsync(
 442        string flowId,
 443        string leaseId,
 444        TimeSpan leaseDuration,
 445        bool acquire,
 446        CancellationToken cancellationToken)
 447    {
 448        DurableFlowStoreShared.ValidateLeaseArgs(flowId, leaseId, leaseDuration);
 449
 450        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 451        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 452        await using var command = connection.CreateCommand();
 453        // Lease fencing runs entirely on the database clock: acquire steals only leases the
 454        // database considers expired, and renew/extend stays relative to now(), so worker clock
 455        // skew can never make two nodes hold the same lease.
 456        command.CommandText =
 457            $"""
 458            UPDATE {Table}
 459            SET lease_id = @lease_id, lease_expires_at_utc = now() + @lease_duration
 460            WHERE flow_id = @flow_id
 461              AND expires_at_utc > now()
 462              AND {(acquire ? "(lease_id IS NULL OR lease_expires_at_utc <= now() OR lease_id = @lease_id)" : "lease_id 
 463            """;
 464        command.Parameters.AddWithValue("flow_id", flowId);
 465        command.Parameters.AddWithValue("lease_id", leaseId);
 466        command.Parameters.Add("lease_duration", NpgsqlDbType.Interval).Value = DurableFlowStoreShared.ServerClockTtl(le
 467        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 468    }
 469
 470    /// <summary>Disposes the data source when the store created (and therefore owns) it.</summary>
 471    public void Dispose()
 472    {
 473        _ensureGate.Dispose();
 474        if (_ownsDataSource)
 475            _dataSource.Dispose();
 476    }
 477
 478    /// <inheritdoc cref="Dispose" />
 479    public async ValueTask DisposeAsync()
 480    {
 481        _ensureGate.Dispose();
 482        if (_ownsDataSource)
 483            await _dataSource.DisposeAsync().ConfigureAwait(false);
 484    }
 485
 486    private string Schema => Quote(_options.SchemaName);
 487    private string Table => $"{Schema}.{Quote(_options.TableName)}";
 488    private string IndexName => Quote(DurableFlowStoreShared.DerivedName(_options.TableName, "_expires_idx", 63));
 489    private static string Quote(string identifier) => "\"" + identifier.Replace("\"", "\"\"", StringComparison.Ordinal) 
 490
 491    /// <summary>
 492    /// A single-quoted SQL string literal, for the catalog lookups inside the DDL's DO block where
 493    /// a parameter cannot be bound.
 494    /// </summary>
 495    private static string SqlLiteral(string value) => "'" + value.Replace("'", "''", StringComparison.Ordinal) + "'";
 496}
 497}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

get_ConnectionString()
get_SchemaName()
get_TableName()
get_AutoCreateSchema()
get_PruneInterval()
get_PruneBudget()
get_MaxStateBytes()
Validate()