Class InMemoryStorage

Namespace

Millrace.Storage.InMemory

Assembly

Millrace.dll

The bundled in-memory provider (ARCHITECTURE.md §4 P5) — for development, samples, and tests. Explicitly not durable. A single lock serializes every operation, which makes the atomicity contract trivially true; the value of this implementation is precision, not speed.

public sealed class InMemoryStorage : IJobStorage, IWorkflowStorage, IStorageNotifier, IMonitoringStorage

Inheritance

object

InMemoryStorage

Implements

IJobStorage

IWorkflowStorage

IStorageNotifier

IMonitoringStorage

Inherited Members

object.Equals(object)

object.Equals(object, object)

object.GetHashCode()

object.GetType()

object.ReferenceEquals(object, object)

object.ToString()

Remarks

Held to exactly the same observable contract as a relational provider — ordering, cursor handling and limit clamping all match, because the conformance kit runs the same facts against both. Where the two could drift, this defers to the shared MonitoringCursor.

Constructors

InMemoryStorage(TimeProvider?)

Creates an empty store.

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.

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.