Table of Contents

Interface IFlowRunStore

Namespace
FlowOrchestrator.Core.Storage
Assembly
FlowOrchestrator.Core.dll

Persistence contract for flow run lifecycle: creating runs, recording step progress, completing runs, and querying run history.

public interface IFlowRunStore

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.

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

Parameters

runId Guid
stepKey string
jobId string
ct CancellationToken

Returns

Task

CompleteRunAsync(Guid, string)

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

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.

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).

GetActiveRunsAsync()

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

Task<IReadOnlyList<FlowRunRecord>> GetActiveRunsAsync()

Returns

Task<IReadOnlyList<FlowRunRecord>>

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.

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.

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

Parameters

runId Guid

Returns

Task<IReadOnlySet<string>>

GetRunDetailAsync(Guid)

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

Task<FlowRunRecord?> GetRunDetailAsync(Guid runId)

Parameters

runId Guid

Returns

Task<FlowRunRecord>

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.

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.

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.

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?).

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.

Task<DashboardStatistics> GetStatisticsAsync()

Returns

Task<DashboardStatistics>

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

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

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: sets status to Running, persists inputs, and associates the Hangfire job ID.

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

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.

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

Parameters

runId Guid
stepKey string
ct CancellationToken

Returns

Task

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.

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.

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.

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>

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).

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)).

Edit this page