Class: InMemoryWorkflowPersistence

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:84

Full persistence contract: PersistenceCore (what the kernel calls) plus a transaction boundary that hands the callback this same surface. The pre-1.0 artifact methods (saveArtifact, loadArtifact, ... — replaced by the BlobStore port) and query helpers (getRunsByStatus, claimPendingRun, getStageById, ...) are no longer part of the contract; the built-in adapters still implement them as plain class methods.

Implements

• WorkflowPersistence

Constructors

Constructor

new InMemoryWorkflowPersistence(opts?): InMemoryWorkflowPersistence

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:109

Parameters

opts?

InMemoryPersistenceOptions = {}

Returns

InMemoryWorkflowPersistence

Methods

acquireIdempotencyKey()

acquireIdempotencyKey(key, commandType, options?): Promise<{ status: "acquired"; } | { result: unknown; status: "replay"; } | { status: "in_progress"; }>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:759

Atomically acquire an idempotency key for command execution.

If the key is currently in_progress (e.g. a previous dispatcher crashed between committing its transaction and calling completeIdempotencyKey), passing staleInProgressAfterMs allows the key to be reclaimed once it has been in progress for at least that long, measured against options.now (defaults to new Date()). Reclaiming is atomic: only one caller wins when multiple dispatchers race to reclaim the same stale key. When staleInProgressAfterMs is omitted, a stuck in_progress key is never reclaimed (matches prior behavior).

Parameters

key

string

commandType

string

options?

now?

Date

staleInProgressAfterMs?

number

Returns

Promise<{ status: "acquired"; } | { result: unknown; status: "replay"; } | { status: "in_progress"; }>

Implementation of

WorkflowPersistence.acquireIdempotencyKey

---

appendAnnotations()

appendAnnotations(inputs): Promise<void>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:868

Append one or more annotations. Designed to be called both standalone (fire-and-forget from external attach) and inside an existing transaction (buffered during stage execution, flushed in the stage-completion transaction) -- called from run.create, external attach, and the stage-completion transactions in job-execute and stage-poll-suspended.

Rows with the same (workflowRunId, key, idempotencyKey) are deduped via the unique constraint; duplicates are silently skipped.

Parameters

inputs

CreateAnnotationInput[]

Returns

Promise<void>

Implementation of

WorkflowPersistence.appendAnnotations

---

appendOutboxEvents()

appendOutboxEvents(events): Promise<void>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:663

Write events to the outbox. Sequences are auto-assigned per workflowRunId.

Parameters

events

CreateOutboxEventInput[]

Returns

Promise<void>

Implementation of

WorkflowPersistence.appendOutboxEvents

---

claimNextPendingRun()

claimNextPendingRun(options?, attempt?): Promise<WorkflowRunRecord | null>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:306

Atomically find and claim the next pending workflow run. Uses FOR UPDATE SKIP LOCKED pattern (in Postgres) to prevent race conditions when multiple workers try to claim workflows simultaneously.

Priority ordering: higher priority first, then oldest (FIFO within same priority).

Parameters

options?

now?

Date

serves?

readonly ServedDefinition[]

attempt?

number = 0

Returns

Promise<WorkflowRunRecord | null>

The claimed workflow run (now with status RUNNING), or null if no pending runs

Implementation of

WorkflowPersistence.claimNextPendingRun

---

claimPendingRun()

claimPendingRun(id): Promise<boolean>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:288

Parameters

id

string

Returns

Promise<boolean>

---

claimUnpublishedOutboxEvents()

claimUnpublishedOutboxEvents(limit?): Promise<OutboxRecord[]>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:698

Atomically claim up to limit unpublished events for this caller by setting publishedAt, ordered by (workflowRunId, sequence). Two processes flushing the same outbox concurrently must never both receive the same event: on Postgres this is one UPDATE ... FROM (SELECT ... FOR UPDATE SKIP LOCKED) RETURNING; other stores use a compare-and-set on publishedAt IS NULL per row. Events whose publication then fails are handed back with releaseOutboxEvents.

Parameters

limit?

number

Returns

Promise<OutboxRecord[]>

Implementation of

WorkflowPersistence.claimUnpublishedOutboxEvents

---

clear()

clear(): void

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:1065

Clear all data - useful between tests

Returns

void

---

completeIdempotencyKey()

completeIdempotencyKey(key, commandType, result): Promise<void>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:792

Mark an idempotency key as completed and cache the command result.

Parameters

key

string

commandType

string

result

unknown

Returns

Promise<void>

Implementation of

WorkflowPersistence.completeIdempotencyKey

---

countRunsByDefinitionVersion()

countRunsByDefinitionVersion(filter?): Promise<DefinitionVersionCount[]>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:1016

Run counts grouped by (workflowId, definitionVersion, status) — the query behind "has this version drained?". Returns an empty array when PersistenceCore.supportsDefinitionVersioning is false.

Parameters

filter?

DefinitionVersionCountFilter

Returns

Promise<DefinitionVersionCount[]>

Implementation of

WorkflowPersistence.countRunsByDefinitionVersion

---

createLog()

createLog(data): Promise<void>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:646

Parameters

data

CreateLogInput

Returns

Promise<void>

Implementation of

WorkflowPersistence.createLog

---

createRun()

createRun(data): Promise<WorkflowRunRecord>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:137

Parameters

data

CreateRunInput

Returns

Promise<WorkflowRunRecord>

Implementation of

WorkflowPersistence.createRun

---

createStage()

createStage(data): Promise<WorkflowStageRecord>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:377

Parameters

data

CreateStageInput

Returns

Promise<WorkflowStageRecord>

Implementation of

WorkflowPersistence.createStage

---

deleteArtifact()

deleteArtifact(runId, key): Promise<void>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:846

Parameters

runId

string

key

string

Returns

Promise<void>

---

deleteRun()

deleteRun(id, cutoff, statuses): Promise<boolean>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:228

Delete a run and everything the persistence owns under it: stage records, logs, artifacts, annotations, and (on the reference schema, through the cascade) workflow_steps. Never emits an outbox event; the kernel clears the step ledger and the blob store before calling it.

Only while the run still matches listRunsForPurge(cutoff, statuses), checked atomically with the delete: a redrive can reopen a listed run at any point of the sweep. Resolves whether the run was deleted, so a missing id or a run no longer eligible is a no-op that resolves false.

Parameters

id

string

cutoff

Date

statuses

readonly PurgeableRunStatus[]

Returns

Promise<boolean>

Implementation of

WorkflowPersistence.deleteRun

---

deleteStage()

deleteStage(id): Promise<void>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:625

Parameters

id

string

Returns

Promise<void>

Implementation of

WorkflowPersistence.deleteStage

---

getAllAnnotations()

getAllAnnotations(): WorkflowAnnotationRecord[]

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:1114

Get all annotations for inspection

Returns

WorkflowAnnotationRecord[]

---

getAllArtifacts()

getAllArtifacts(): WorkflowArtifactRecord[]

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:1107

Get all artifacts for inspection

Returns

WorkflowArtifactRecord[]

---

getAllLogs()

getAllLogs(): WorkflowLogRecord[]

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:1100

Get all logs for inspection

Returns

WorkflowLogRecord[]

---

getAllRuns()

getAllRuns(): WorkflowRunRecord[]

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:1083

Get all runs for inspection

Returns

WorkflowRunRecord[]

---

getAllStages()

getAllStages(): WorkflowStageRecord[]

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:1090

Get all stages for inspection

Returns

WorkflowStageRecord[]

---

getDefinition()

getDefinition(workflowId, version): Promise<WorkflowDefinitionRecord | null>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:1008

Loads one stored definition snapshot, or null when absent.

Parameters

workflowId

string

version

string

Returns

Promise<WorkflowDefinitionRecord | null>

Implementation of

WorkflowPersistence.getDefinition

---

getFirstFailedStage()

getFirstFailedStage(runId): Promise<WorkflowStageRecord | null>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:596

Parameters

runId

string

Returns

Promise<WorkflowStageRecord | null>

---

getFirstSuspendedStageReadyToResume()

getFirstSuspendedStageReadyToResume(runId): Promise<WorkflowStageRecord | null>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:585

Parameters

runId

string

Returns

Promise<WorkflowStageRecord | null>

---

getLastCompletedStage()

getLastCompletedStage(runId): Promise<WorkflowStageRecord | null>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:603

Parameters

runId

string

Returns

Promise<WorkflowStageRecord | null>

---

getLastCompletedStageBefore()

getLastCompletedStageBefore(runId, executionGroup): Promise<WorkflowStageRecord | null>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:613

Parameters

runId

string

executionGroup

number

Returns

Promise<WorkflowStageRecord | null>

---

getRun()

getRun(id): Promise<WorkflowRunRecord | null>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:192

Parameters

id

string

Returns

Promise<WorkflowRunRecord | null>

Implementation of

WorkflowPersistence.getRun

---

getRunsByStatus()

getRunsByStatus(status): Promise<WorkflowRunRecord[]>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:202

Parameters

status

Status

Returns

Promise<WorkflowRunRecord[]>

---

getRunStatus()

getRunStatus(id): Promise<Status | null>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:197

Parameters

id

string

Returns

Promise<Status | null>

Implementation of

WorkflowPersistence.getRunStatus

---

getStage()

getStage(runId, stageId): Promise<WorkflowStageRecord | null>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:493

Parameters

runId

string

stageId

string

Returns

Promise<WorkflowStageRecord | null>

Implementation of

WorkflowPersistence.getStage

---

getStageById()

getStageById(id): Promise<WorkflowStageRecord | null>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:502

Parameters

id

string

Returns

Promise<WorkflowStageRecord | null>

---

getStageIdForArtifact()

getStageIdForArtifact(runId, stageId): Promise<string | null>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:856

Parameters

runId

string

stageId

string

Returns

Promise<string | null>

---

getStagesByRun()

getStagesByRun(runId, options?): Promise<WorkflowStageRecord[]>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:507

Ordered by executionGroup (actual execution/dependency order), with stageNumber (definition order) as a tiebreaker for stages that share an execution group (parallel stages).

Parameters

runId

string

options?

orderBy?

"asc" | "desc"

status?

Status

Returns

Promise<WorkflowStageRecord[]>

Implementation of

WorkflowPersistence.getStagesByRun

---

getStuckRuns()

getStuckRuns(stuckSince): Promise<WorkflowRunRecord[]>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:267

Parameters

stuckSince

Date

Returns

Promise<WorkflowRunRecord[]>

Implementation of

WorkflowPersistence.getStuckRuns

---

getSuspendedStages()

getSuspendedStages(beforeDate, options?): Promise<WorkflowStageRecord[]>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:536

Suspended stages whose nextPollAt has come due, oldest deadline first, capped at limit. With includeSettlements, also return COMPLETED and FAILED stages with a due deadline, retained for job acknowledgement.

Both narrowings belong here rather than in the caller. Without the ordering and the cap the adapter returns every ready row and the handler slices in JS, so a row that keeps coming back — a stage this build cannot serve, released for another host to take — permanently occupies a candidate slot and pushes servable stages out of the window. serves is the same filter claimNextPendingRun applies to runs, evaluated against the stage's run, so an unserving host does not even claim the poll lease of a stage that is not its work. An adapter whose schema has no definitionVersion column ignores serves; such a database has no pinned runs, so ignoring it is exact.

Parameters

beforeDate

Date

options?

includeSettlements?

boolean

limit?

number

serves?

readonly ServedDefinition[]

Returns

Promise<WorkflowStageRecord[]>

Implementation of

WorkflowPersistence.getSuspendedStages

---

getUnpublishedOutboxEvents()

getUnpublishedOutboxEvents(limit?): Promise<OutboxRecord[]>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:685

Read unpublished events ordered by (workflowRunId, sequence).

Parameters

limit?

number

Returns

Promise<OutboxRecord[]>

Implementation of

WorkflowPersistence.getUnpublishedOutboxEvents

---

hasArtifact()

hasArtifact(runId, key): Promise<boolean>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:842

Parameters

runId

string

key

string

Returns

Promise<boolean>

---

incrementOutboxRetryCount()

incrementOutboxRetryCount(id): Promise<number>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:731

Increment retry count for a failed outbox event. Returns new count.

Parameters

id

string

Returns

Promise<number>

Implementation of

WorkflowPersistence.incrementOutboxRetryCount

---

insertDefinitionIfAbsent()

insertDefinitionIfAbsent(input): Promise<WorkflowDefinitionRecord | null>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:991

Stores a definition snapshot if (workflowId, version) is not already present, and returns the stored row either way — so a caller can tell whether an explicit version is being re-registered with a different structure. Returns null when PersistenceCore.supportsDefinitionVersioning is false.

Parameters

input

CreateDefinitionInput

Returns

Promise<WorkflowDefinitionRecord | null>

Implementation of

WorkflowPersistence.insertDefinitionIfAbsent

---

listAnnotations()

listAnnotations(workflowRunId, filters?): Promise<WorkflowAnnotationRecord[]>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:902

List annotations for a run, optionally filtered. Returns rows ordered by createdAt ascending (so consumers get a timeline by default).

Parameters

workflowRunId

string

filters?

AnnotationFilters = {}

Returns

Promise<WorkflowAnnotationRecord[]>

Implementation of

WorkflowPersistence.listAnnotations

---

listArtifacts()

listArtifacts(runId): Promise<WorkflowArtifactRecord[]>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:850

Parameters

runId

string

Returns

Promise<WorkflowArtifactRecord[]>

---

listRunsForPurge()

listRunsForPurge(cutoff, statuses, limit): Promise<PurgeableRun[]>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:208

Terminal runs eligible for retention deletion, oldest first: runs whose status is one of statuses and which finished at or before cutoff (completedAt <= cutoff, or updatedAt <= cutoff for a terminal run with no completedAt). Returns at most limit runs, each with the ids of its stage records so the caller can clear a pluggable StepLedger before the row goes. Called by run.purge.

Parameters

cutoff

Date

statuses

readonly PurgeableRunStatus[]

limit

number

Returns

Promise<PurgeableRun[]>

Implementation of

WorkflowPersistence.listRunsForPurge

---

loadArtifact()

loadArtifact(runId, key): Promise<unknown>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:837

Parameters

runId

string

key

string

Returns

Promise<unknown>

---

markOutboxEventsPublished()

markOutboxEventsPublished(ids): Promise<void>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:722

Mark events as published.

Parameters

ids

string[]

Returns

Promise<void>

Implementation of

WorkflowPersistence.markOutboxEventsPublished

---

moveOutboxEventToDLQ()

moveOutboxEventToDLQ(id): Promise<void>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:738

Move an outbox event to DLQ (sets dlqAt).

Parameters

id

string

Returns

Promise<void>

Implementation of

WorkflowPersistence.moveOutboxEventToDLQ

---

releaseIdempotencyKey()

releaseIdempotencyKey(key, commandType): Promise<void>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:807

Delete an idempotency key whatever its state: an in-progress key after a command failed, or a completed one nothing will replay any more.

Parameters

key

string

commandType

string

Returns

Promise<void>

Implementation of

WorkflowPersistence.releaseIdempotencyKey

---

releaseOutboxEvents()

releaseOutboxEvents(ids): Promise<void>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:715

Un-claim events (clear publishedAt) so a later flush retries them.

Parameters

ids

string[]

Returns

Promise<void>

Implementation of

WorkflowPersistence.releaseOutboxEvents

---

replayDLQEvents()

replayDLQEvents(maxEvents): Promise<number>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:744

Reset DLQ events so they can be reprocessed by outbox.flush. Returns count reset.

Parameters

maxEvents

number

Returns

Promise<number>

Implementation of

WorkflowPersistence.replayDLQEvents

---

saveArtifact()

saveArtifact(data): Promise<void>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:817

Parameters

data

SaveArtifactInput

Returns

Promise<void>

---

saveStageOutput()

saveStageOutput(runId, workflowType, stageId, output): Promise<string>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:962

Parameters

runId

string

workflowType

string

stageId

string

output

unknown

Returns

Promise<string>

---

supportsDefinitionVersioning()

supportsDefinitionVersioning(): boolean

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:987

Whether this adapter's schema carries the definition-versioning columns and table. false on a database that has not been migrated: the engine then behaves exactly as it did before versioning existed rather than failing to start.

Returns

boolean

Implementation of

WorkflowPersistence.supportsDefinitionVersioning

---

updateRun()

updateRun(id, data): Promise<void>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:165

Updates a run's fields. version is incremented on every call, whether or not expectedVersion is supplied -- callers relying on optimistic concurrency (e.g. job.execute's claimed-run guard) can always detect a concurrent write, including unconditional writes like run.cancel.

Parameters

id

string

data

UpdateRunInput

Returns

Promise<void>

Implementation of

WorkflowPersistence.updateRun

---

updateStage()

updateStage(id, data): Promise<void>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:432

Parameters

id

string

data

UpdateStageInput

Returns

Promise<void>

Implementation of

WorkflowPersistence.updateStage

---

updateStageByRunAndStageId()

updateStageByRunAndStageId(workflowRunId, stageId, data): Promise<void>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:460

Parameters

workflowRunId

string

stageId

string

data

UpdateStageInput

Returns

Promise<void>

---

upsertStage()

upsertStage(data): Promise<WorkflowStageRecord>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:414

Parameters

data

UpsertStageInput

Returns

Promise<WorkflowStageRecord>

Implementation of

WorkflowPersistence.upsertStage

---

withTransaction()

withTransaction<T>(fn): Promise<T>

Defined in: packages/workflow-engine/src/testing/in-memory-persistence.ts:127

Execute operations within a transaction boundary. Redeclared (not merely inherited from PersistenceCore) so the callback receives the full WorkflowPersistence surface.

Type Parameters

T

T

Parameters

fn

(tx) => Promise<T>

Returns

Promise<T>

Implementation of

WorkflowPersistence.withTransaction