Table of Contents

Class PostgreSqlFlowRunStore

Namespace
FlowOrchestrator.PostgreSQL
Assembly
FlowOrchestrator.PostgreSQL.dll

Dapper-based PostgreSQL implementation of all run storage interfaces. Uses PostgreSQL's INSERT ... ON CONFLICT DO NOTHING for atomic step claim deduplication.

public sealed class PostgreSqlFlowRunStore : IFlowRunStore, IFlowRunRuntimeStore, IFlowRunControlStore, IFlowRetentionStore
Inheritance
PostgreSqlFlowRunStore
Implements
Inherited Members

Constructors

PostgreSqlFlowRunStore(string) 

public PostgreSqlFlowRunStore(string connectionString)

Parameters

connectionString string

Methods

AnnotateDispatchAsync(Guid, string, string, CancellationToken)

Stores the runtime job or message ID returned by the dispatcher alongside the dispatch record. Best-effort — implementations should not throw; failures are silently ignored by the engine.

public Task AnnotateDispatchAsync(Guid runId, string stepKey, string jobId, CancellationToken ct = default)

Parameters

runId Guid
stepKey string
jobId string
ct CancellationToken

Returns

Task

CleanupAsync(DateTimeOffset, CancellationToken)

Deletes all run data (runs, steps, attempts, outputs, events) whose completion time is older than cutoffUtc.

public Task CleanupAsync(DateTimeOffset cutoffUtc, CancellationToken cancellationToken)

Parameters

cutoffUtc DateTimeOffset

Runs completed before this timestamp are eligible for deletion.

cancellationToken CancellationToken

Propagates cancellation from the host shutdown signal.

Returns

Task

CompleteRunAsync(Guid, string)

Marks the run as complete (status: Succeeded, Failed, or Cancelled) and sets CompletedAt.

public Task CompleteRunAsync(Guid runId, string status)

Parameters

runId Guid
status string

Returns

Task

CompleteRunIfActiveAsync(Guid, string)

Transitions the run to a terminal status only if it is still active (Running), reporting whether this call performed the transition. Makes run completion idempotent and safe against concurrent completers — notably the graph continuation and the periodic timeout sweep (single- or multi-instance) — so a run's lifecycle event is published exactly once and its terminal status is set by exactly one writer.

public Task<bool> CompleteRunIfActiveAsync(Guid runId, string status)

Parameters

runId Guid

The run to complete.

status string

The terminal status to set.

Returns

Task<bool>

true if this call transitioned a Running run to status; false if the run was already in a terminal state (another writer won).

Remarks

Default implementation delegates to CompleteRunAsync(Guid, string) and reports true, so existing custom IFlowRunStore implementations compile unchanged (retaining their prior, non-idempotent completion behaviour).

ConfigureRunAsync(Guid, Guid, string, string?, DateTimeOffset?)

Persists the control record for a new run, including an optional idempotency key and an absolute timeout deadline.

public Task ConfigureRunAsync(Guid runId, Guid flowId, string triggerKey, string? idempotencyKey, DateTimeOffset? timeoutAtUtc)

Parameters

runId Guid
flowId Guid
triggerKey string
idempotencyKey string
timeoutAtUtc DateTimeOffset?

Returns

Task

ExtendDeadlineAsync(Guid, DateTimeOffset?)

Grants a run a fresh execution window: sets TimeoutAtUtc to newTimeoutAtUtc (or clears the deadline when null) and un-latches any timeout-induced termination — clears TimedOutAtUtc and the cancellation fields that MarkTimedOutAsync(Guid, string?) set. A genuine user cancellation (recorded via RequestCancelAsync(Guid, string?) while the run had not timed out) is preserved.

public Task<bool> ExtendDeadlineAsync(Guid runId, DateTimeOffset? newTimeoutAtUtc)

Parameters

runId Guid

The run whose deadline is being refreshed.

newTimeoutAtUtc DateTimeOffset?

The new absolute deadline, or null to leave the run without a timeout bound.

Returns

Task<bool>

true if the record was found and updated; false otherwise.

Remarks

Called by FlowOrchestratorEngine.RetryStepAsync before re-dispatch so a step retried after the run's deadline lapsed can actually re-execute instead of being skipped by the termination gate. Default implementation is a no-op returning false so existing custom IFlowRunControlStore implementations continue to compile.

FindRunIdByIdempotencyKeyAsync(Guid, string, string)

Looks up an existing run that was started with the given idempotency key. Returns the RunId of the existing run, or null if none exists.

public Task<Guid?> FindRunIdByIdempotencyKeyAsync(Guid flowId, string triggerKey, string idempotencyKey)

Parameters

flowId Guid
triggerKey string
idempotencyKey string

Returns

Task<Guid?>

GetActiveRunsAsync()

Returns all runs currently in Running status (used for timeout enforcement).

public Task<IReadOnlyList<FlowRunRecord>> GetActiveRunsAsync()

Returns

Task<IReadOnlyList<FlowRunRecord>>

GetClaimedStepKeysAsync(Guid)

Returns the set of step keys that have been claimed (locked) for execution but not yet completed, used to detect in-progress steps.

public Task<IReadOnlyCollection<string>> GetClaimedStepKeysAsync(Guid runId)

Parameters

runId Guid

Returns

Task<IReadOnlyCollection<string>>

GetDerivedRunsAsync(Guid)

Returns runs whose SourceRunId equals sourceRunId, i.e. all re-runs derived from a given run. Used by the dashboard to render lineage.

public Task<IReadOnlyList<FlowRunRecord>> GetDerivedRunsAsync(Guid sourceRunId)

Parameters

sourceRunId Guid

Returns

Task<IReadOnlyList<FlowRunRecord>>

GetDispatchedStepKeysAsync(Guid)

Returns the set of step keys that have been dispatched (and not yet released) for a run. Used by the recovery service to avoid re-dispatching already-in-flight steps.

public Task<IReadOnlySet<string>> GetDispatchedStepKeysAsync(Guid runId)

Parameters

runId Guid

Returns

Task<IReadOnlySet<string>>

GetRunControlAsync(Guid)

Returns the control record for the given run, or null if not found.

public Task<FlowRunControlRecord?> GetRunControlAsync(Guid runId)

Parameters

runId Guid

Returns

Task<FlowRunControlRecord>

GetRunDetailAsync(Guid)

Returns full run detail including all step records and their attempt history, or null if no run with runId exists.

public Task<FlowRunRecord?> GetRunDetailAsync(Guid runId)

Parameters

runId Guid

Returns

Task<FlowRunRecord>

GetRunStatusAsync(Guid)

Returns the current overall status of the run ("Running", "Succeeded", etc.), or null if the run does not exist.

public Task<string?> GetRunStatusAsync(Guid runId)

Parameters

runId Guid

Returns

Task<string>

GetRunTimeseriesAsync(RunTimeseriesGranularity, DateTimeOffset, DateTimeOffset, Guid?)

Returns time-bucketed run counts and duration percentiles for the half-open interval [since, until). Buckets that contain no runs are still returned (with zero counts) so the caller can render a contiguous timeline without gap-filling. When flowId is supplied, only runs for that flow are included.

public Task<IReadOnlyList<RunTimeseriesBucket>> GetRunTimeseriesAsync(RunTimeseriesGranularity granularity, DateTimeOffset since, DateTimeOffset until, Guid? flowId = null)

Parameters

granularity RunTimeseriesGranularity

Hour or Day bucket size.

since DateTimeOffset

UTC start of the window (inclusive). Aligned to the granularity boundary.

until DateTimeOffset

UTC end of the window (exclusive). Aligned to the granularity boundary.

flowId Guid?

Optional filter — when non-null, only runs for this flow are counted.

Returns

Task<IReadOnlyList<RunTimeseriesBucket>>

GetRunsAsync(Guid?, int, int)

Returns a page of run records, optionally filtered by flowId. Results are ordered by start time descending.

public Task<IReadOnlyList<FlowRunRecord>> GetRunsAsync(Guid? flowId = null, int skip = 0, int take = 50)

Parameters

flowId Guid?
skip int
take int

Returns

Task<IReadOnlyList<FlowRunRecord>>

GetRunsPageAsync(Guid?, string?, int, int, string?)

Returns a paginated run list with total count, optionally filtered by flow, status, and free-text search.

public Task<(IReadOnlyList<FlowRunRecord> Runs, int TotalCount)> GetRunsPageAsync(Guid? flowId = null, string? status = null, int skip = 0, int take = 50, string? search = null)

Parameters

flowId Guid?

Restricts results to a single flow when set.

status string

Restricts results to runs in the given status when set.

skip int

Number of leading rows to skip (offset pagination).

take int

Maximum number of rows to return.

search string

Free-text term matched (case-insensitively) against run identity columns and the current step records (step key, error message, output JSON). Step attempt history is intentionally not searched — its output duplicates the current step row.

Returns

Task<(IReadOnlyList<FlowRunRecord> Runs, int TotalCount)>

GetRunsPageAsync(Guid?, string?, int, int, string?, bool, DateTimeOffset?, DateTimeOffset?)

Tiered, time-bounded variant of GetRunsPageAsync(Guid?, string?, int, int, string?).

public Task<(IReadOnlyList<FlowRunRecord> Runs, int TotalCount)> GetRunsPageAsync(Guid? flowId, string? status, int skip, int take, string? search, bool deepSearch, DateTimeOffset? startedFrom = null, DateTimeOffset? startedTo = null)

Parameters

flowId Guid?
status string
skip int
take int
search string
deepSearch bool
startedFrom DateTimeOffset?
startedTo DateTimeOffset?

Returns

Task<(IReadOnlyList<FlowRunRecord> Runs, int TotalCount)>

Remarks

deepSearch: true (the behaviour of the five-argument overload) also matches the current step records; deepSearch: false matches only the top-level run columns (id, flow name, trigger key, status, background job id) — index-friendlier, suited to typeahead such as the command palette. startedFrom / startedTo bound the scan to a start-time window when set. The default implementation delegates to the five-argument overload, so existing IFlowRunStore providers keep compiling and stay correct (they simply do not get the quick-search or time-window optimisations until they override this method).

GetStatisticsAsync()

Returns aggregate counts used by the dashboard overview panel.

public Task<DashboardStatistics> GetStatisticsAsync()

Returns

Task<DashboardStatistics>

GetStepStatusesAsync(Guid)

Returns the current StepStatus for every step in the run.

public Task<IReadOnlyDictionary<string, StepStatus>> GetStepStatusesAsync(Guid runId)

Parameters

runId Guid

Returns

Task<IReadOnlyDictionary<string, StepStatus>>

MarkTimedOutAsync(Guid, string?)

Marks the run as timed out.

public Task<bool> MarkTimedOutAsync(Guid runId, string? reason)

Parameters

runId Guid
reason string

Returns

Task<bool>

true if the record was found and updated; false otherwise.

Remarks

Invoked from two places: lazily by the engine when a step is dispatched after the deadline has passed, and proactively by the periodic timeout-enforcement hosted service (FlowTimeoutEnforcementHostedService). Also sets the cancellation latch so in-flight steps short-circuit on their next dispatch — a fresh execution window can be granted afterwards via ExtendDeadlineAsync(Guid, DateTimeOffset?).

RecordSkippedStepAsync(Guid, string, string, string?)

Records a step as Skipped without executing it, used when runAfter conditions cannot be satisfied.

public Task RecordSkippedStepAsync(Guid runId, string stepKey, string stepType, string? reason)

Parameters

runId Guid
stepKey string
stepType string
reason string

Returns

Task

RecordSkippedStepAsync(Guid, string, string, string?, string?)

Records a Skipped step and persists the When-clause evaluation trace onto the current step row and its latest attempt.

public Task RecordSkippedStepAsync(Guid runId, string stepKey, string stepType, string? reason, string? evaluationTraceJson)

Parameters

runId Guid
stepKey string
stepType string
reason string
evaluationTraceJson string

Returns

Task

Remarks

Overrides the runtime-store default so PostgreSQL surfaces the "Why skipped" trace in run detail, matching the SQL Server backend.

RecordStepCompleteAsync(Guid, string, string, string?, string?)

Records the outcome of a step attempt: updates status, persists output JSON, and sets error message on failure.

public Task RecordStepCompleteAsync(Guid runId, string stepKey, string status, string? outputJson, string? errorMessage)

Parameters

runId Guid
stepKey string
status string
outputJson string
errorMessage string

Returns

Task

RecordStepStartAsync(Guid, string, string, string?, string?)

Records the start of a step attempt, allocating the next attempt number while holding a transaction-scoped advisory lock keyed on the step so concurrent attempts serialise.

public Task RecordStepStartAsync(Guid runId, string stepKey, string stepType, string? inputJson, string? jobId)

Parameters

runId Guid
stepKey string
stepType string
inputJson string
jobId string

Returns

Task

Remarks

SQL Server serialises the MAX(attempt_no)+1 allocation with UPDLOCK, HOLDLOCK — it blocks the second writer rather than aborting it. Postgres has no equivalent row-range hint for this read-then-insert, and a plain Serializable transaction surfaces the concurrent collision as an error (either a 40001 serialization failure or, more commonly here, a 23505 unique-violation on flow_step_attempts_pkey when both transactions compute the same attempt number). To match SQL Server's non-throwing semantics, this method takes a pg_advisory_xact_lock keyed on (run_id, step_key) at the top of the transaction: concurrent callers for the same step block until the holder commits, so the next MAX(attempt_no) read always sees prior attempts. The lock is released automatically when the transaction commits or rolls back.

ReleaseDispatchAsync(Guid, string, CancellationToken)

Removes the dispatch record for a step, allowing it to be re-dispatched. Called by the engine before rescheduling a Pending (polling) step.

public Task ReleaseDispatchAsync(Guid runId, string stepKey, CancellationToken ct = default)

Parameters

runId Guid
stepKey string
ct CancellationToken

Returns

Task

ReleaseStepClaimAsync(Guid, string)

Releases a previously-acquired step claim so a future TryClaimStepAsync(Guid, string) for the same (runId, stepKey) can succeed. Called by the engine on Pending re-schedule and retry paths, where the same logical step needs to run again.

public Task ReleaseStepClaimAsync(Guid runId, string stepKey)

Parameters

runId Guid
stepKey string

Returns

Task

Remarks

Idempotent — calling for a key with no claim is a no-op. Default implementation is a no-op so existing custom runtime stores continue to compile; in that case retry/Pending paths behave as before v1.22 (the schedule-time claim was non-strict). New implementations should remove the claim row atomically.

RequestCancelAsync(Guid, string?)

Marks a cancellation request for the run. Steps check this flag before executing.

public Task<bool> RequestCancelAsync(Guid runId, string? reason)

Parameters

runId Guid
reason string

Returns

Task<bool>

true if the record was found and updated; false otherwise.

ResetCascadeSkippedDependentsAsync(Guid, IReadOnlyCollection<string>)

Clears cascade-skip records (PrerequisitesUnmet) for the given step keys so the DAG planner re-evaluates them after a manual retry of an upstream step. Removes the step row, claim row, and dispatch-ledger row for any matching descendant — the attempt history (FlowStepAttempts) is preserved so the dashboard still surfaces the prior "Skipped" attempt as audit trail.

public Task ResetCascadeSkippedDependentsAsync(Guid runId, IReadOnlyCollection<string> stepKeys)

Parameters

runId Guid
stepKeys IReadOnlyCollection<string>

Returns

Task

Remarks

Only rows whose status is Skipped AND whose error message exactly matches PrerequisitesUnmet are cleared. Steps skipped via When-clause evaluation carry a different reason and are intentionally left untouched. Idempotent — calling with keys that have no matching row is a no-op. Default implementation is a no-op so external IFlowRunStore implementations continue to compile; in that case the post-retry DAG advance behaves as pre-fix (downstream steps stay Skipped).

RetryStepAsync(Guid, string)

Resets a step back to a re-runnable state so it can be re-enqueued by the retry flow. Clears Running/Failed status and increments attempt count.

public Task RetryStepAsync(Guid runId, string stepKey)

Parameters

runId Guid
stepKey string

Returns

Task

StartRunAsync(Guid, string, Guid, string, string?, string?, Guid?)

Creates a new run record in Running status and returns it. Called once per TriggerAsync invocation.

public Task<FlowRunRecord> StartRunAsync(Guid flowId, string flowName, Guid runId, string triggerKey, string? triggerData, string? jobId, Guid? sourceRunId = null)

Parameters

flowId Guid

The flow definition this run belongs to.

flowName string

Display name of the flow at trigger time.

runId Guid

Unique identifier for this run, generated by the engine.

triggerKey string

The manifest trigger key that fired (e.g. "manual", "schedule").

triggerData string

JSON-serialised trigger payload, or null.

jobId string

Runtime job ID (e.g. Hangfire BackgroundJobId) for cross-referencing, or null.

sourceRunId Guid?

When this run was created via "Re-run all" on a previous run, the ID of that run; otherwise null.

Returns

Task<FlowRunRecord>

TryClaimStepAsync(Guid, string)

Atomically claims a step for execution, returning true if this caller acquired the claim or false if another worker already claimed it.

public Task<bool> TryClaimStepAsync(Guid runId, string stepKey)

Parameters

runId Guid
stepKey string

Returns

Task<bool>

Remarks

This is the primary guard against duplicate step execution. Since v1.22 the engine calls this at the top of RunStepAsync (execute time) rather than at schedule time, so it correctly prevents concurrent execution under at-least-once delivery — including the Service Bus topic-broadcast case where a single dispatched message reaches multiple subscriptions. Implementations must use an atomic compare-and-set or equivalent database primitive. Pair with ReleaseStepClaimAsync(Guid, string) on retry / Pending re-schedule paths so the SAME step can claim again on a fresh attempt.

TryRecordDispatchAsync(Guid, string, CancellationToken)

Atomically records that a step has been dispatched for execution. Returns true if this is the first dispatch for this step in this run; false if the step was already dispatched (idempotent guard — caller should skip).

public Task<bool> TryRecordDispatchAsync(Guid runId, string stepKey, CancellationToken ct = default)

Parameters

runId Guid
stepKey string
ct CancellationToken

Returns

Task<bool>

Remarks

Must use an atomic INSERT-if-not-exists primitive (SQL WHERE NOT EXISTS, PostgreSQL ON CONFLICT DO NOTHING, or TryAdd(TKey, TValue)).

TryRegisterIdempotencyKeyAsync(Guid, string, string, Guid)

Atomically registers an idempotency key for the given run.

public Task<bool> TryRegisterIdempotencyKeyAsync(Guid flowId, string triggerKey, string idempotencyKey, Guid runId)

Parameters

flowId Guid
triggerKey string
idempotencyKey string
runId Guid

Returns

Task<bool>

true if the key was registered by this call; false if a record with this key already existed (duplicate trigger).

Edit this page