Table of Contents

Class PostgreSqlStorage

Namespace
Millrace.Storage.PostgreSql
Assembly
Millrace.Storage.PostgreSql.dll

The PostgreSQL reference provider (ARCHITECTURE.md §11.2): claims via FOR UPDATE SKIP LOCKED, idempotency scopes via a partial unique index with NULLS NOT DISTINCT (PostgreSQL 15+), cancel cascades via a recursive CTE, and push wakeups via LISTEN/NOTIFY. Every now comparison uses the injected TimeProvider, never database time, so the conformance kit drives this provider with a fake clock exactly like InMemory.

public sealed class PostgreSqlStorage : IJobStorage, IWorkflowStorage, IStorageNotifier, IMonitoringStorage
Inheritance
PostgreSqlStorage
Implements
Inherited Members

Remarks

Reads are plain SELECTs taking no locks, so a dashboard cannot delay claiming or applying a transition. Keyset paging uses PostgreSQL's row-value comparison, which the (created_at DESC, id DESC) indexes serve directly — so page 900 costs the same as page 1, which is the point of §11.12.

Constructors

PostgreSqlStorage(NpgsqlDataSource, TimeProvider?, PostgreSqlStorageOptions?)

Creates the provider over an existing data source.

Properties

Capabilities

Optional powers this provider offers, which the engine adapts to.

Methods

ActivateDueJobsAsync(DateTimeOffset, int, CancellationToken)

Moves due Scheduled/Failed jobs (DueAt <= now) to Enqueued, clearing DueAt, in DueAt ASC order (oldest first; ties by enqueue order), up to batchSize; returns the number moved. Safe to run concurrently on every node — each job activates exactly once.

AddBookmarkAsync(BookmarkRecord, CancellationToken)

Records that an instance is waiting for a signal.

ApplyAsync(JobTransition, CancellationToken)

Applies an engine-computed transition atomically behind the fence (§4.2.3); false = fence rejected, nothing changed.

ClaimAsync(ClaimRequest, CancellationToken)

Exclusively claims up to MaxCount jobs (§4.2.1–2).

ConsumeBookmarkAsync(string, string, CancellationToken)

Atomically consumes (removes and returns) the oldest matching bookmark — ordered by CreatedAt, then Id — or returns null when none match. At-most-once under arbitrary concurrency (§4.2.4): a signal resumes exactly one waiting instance.

CreateInstanceAsync(WorkflowInstanceRecord, CancellationToken)

Stores a new instance with Revision = 1; duplicate id throws MillraceConcurrencyException.

EnqueueAsync(IReadOnlyList<JobRecord>, CancellationToken)

Inserts jobs all-or-nothing and returns their effective ids positionally.

GetDueRecurringAsync(DateTimeOffset, int, CancellationToken)

Plain read of records with NextFireTime <= now, ordered NextFireTime ASC (most overdue first, so a backlog cannot starve old definitions), up to batchSize.

GetInstanceAsync(WorkflowInstanceId, CancellationToken)

Reads one instance, or null if it does not exist.

GetJobAsync(JobId, CancellationToken)

Reads one job, or null if no such job exists.

GetJobDetailsAsync(JobId, CancellationToken)

Full detail for one job, or null if no such job exists.

GetRecurringAsync(string, CancellationToken)

Reads one recurring definition, or null if none is registered under that id.

GetStatisticsAsync(TenantFilter, CancellationToken)

Aggregate counts for the overview, scoped by tenant.

InitializeAsync(CancellationToken)

Creates the schema and tables (idempotent). Called lazily unless disabled.

ListenAsync(IReadOnlySet<string>, CancellationToken)

Streams wakeup hints for queues until cancelled.

QueryInstancesAsync(InstanceQuery, CancellationToken)

One page of workflow instances matching query. Same paging, cursor and tenancy rules as QueryJobsAsync(JobQuery, CancellationToken).

QueryJobsAsync(JobQuery, CancellationToken)

One page of jobs matching query.

QueryRecurringAsync(RecurringQuery, CancellationToken)

One page of recurring definitions matching query, ordered NextFireTime ASC, Id ASC.

RemoveRecurringAsync(string, CancellationToken)

Removes a recurring definition.

RenewLeasesAsync(string, IReadOnlyList<JobId>, TimeSpan, CancellationToken)

Extends leases for in-flight jobs; returns the ids actually renewed.

TryCancelAsync(JobId, CancellationToken)

Cancels a job. Atomic: Scheduled/Enqueued/Failed/Awaiting ⇒ Cancelled (FinishedAt set, key released, transitive Awaiting-descendant cascade) and returns true; Processing ⇒ sets CancelRequested only (state and fence untouched) and returns true; terminal or unknown ⇒ false, no mutation. CancelRequested never blocks a fenced ApplyAsync(JobTransition, CancellationToken) — a completing worker may still win with Succeeded.

TryFireRecurringAsync(string, DateTimeOffset, DateTimeOffset, JobRecord, CancellationToken)

Fenced fire (§4.2.5, strengthened): compare-and-set on (id, expectedFireTime) advancing NextFireTime to nextFireTime and setting LastFireTime = expected, inserting job in the same atomic operation iff the CAS wins. Returns whether this caller won — exactly one node enqueues each occurrence, with no crash window between fence and enqueue.

TryRunNowAsync(JobId, CancellationToken)

Makes a job that is waiting out its retry backoff claimable immediately (§11.32). Atomic: Failed ⇒ Enqueued with DueAt cleared, returning true; any other state, or unknown, ⇒ false with no mutation.

UpdateInstanceAsync(WorkflowInstanceRecord, long, CancellationToken)

Optimistic-concurrency replace: throws MillraceConcurrencyException unless the stored revision equals expectedRevision (a missing instance is the same failure — providers must not distinguish); on success stores Revision = expectedRevision + 1.

UpsertRecurringAsync(RecurringJobRecord, CancellationToken)

Single atomic upsert. Insert stores the record as given. Update overwrites Cron/Queue/Invocation/Retry/Priority/TenantId/UpdatedAt; takes NextFireTime from the record iff the stored Cron differs from the record's (else preserves the stored value — the engine always passes a freshly computed NextFireTime, conditionally unused); always preserves LastFireTime/CreatedAt.

Edit this page