Class: InMemoryJobQueue
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:108
Implements
Constructors
Constructor
new InMemoryJobQueue(
workerIdOrOpts?,maybeOpts?):InMemoryJobQueue
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:135
Parameters
workerIdOrOpts?
string | InMemoryJobQueueOptions
Either a worker id string (backwards compatible with the original single-argument constructor) or an options object.
maybeOpts?
InMemoryJobQueueOptions
Options, only consulted when the first argument is a worker id string.
Returns
InMemoryJobQueue
Methods
adoptWorkerId()
adoptWorkerId(
workerId):string
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:158
Take the host's worker id unless this queue was constructed with one of its own; returns the id it will stamp on claimed jobs.
Parameters
workerId
string
Returns
string
Implementation of
cancelByRun()
cancelByRun(
workflowRunId):Promise<number>
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:600
Cancel all pending/suspended jobs for a workflow run. Returns count of cancelled jobs.
Parameters
workflowRunId
string
Returns
Promise<number>
Implementation of
clear()
clear():
void
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:628
Clear all jobs - useful between tests
Returns
void
complete()
complete(
jobId,fence?):Promise<JobAckOutcome>
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:349
Mark job as completed.
Passing a fence conditions the write on the job still being the RUNNING
attempt with that startedAt; omitting it keeps the previous unconditional
behaviour and always returns "acknowledged". Note that the fenced form is
the recommended one and that the unfenced form exists for transports that
cannot carry the stamp.
Parameters
jobId
string
fence?
Returns
Promise<JobAckOutcome>
Implementation of
defer()
defer(
jobId,nextPollAt,reason,fence?):Promise<JobAckOutcome>
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:428
Return a claimed job to PENDING with a later nextPollAt without
counting the claim as an attempt. See JobQueue.defer.
Parameters
jobId
string
nextPollAt
Date
reason
string
fence?
Returns
Promise<JobAckOutcome>
Implementation of
deleteByRunAndStages()
deleteByRunAndStages(
workflowRunId,stageIds):Promise<number>
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:227
Remove every job row for the given stages of a run, whatever their status. Returns the number of rows removed.
Called by run.rerunFrom for the stage records it deletes, so a
rerun does not leave the retired stages' job rows behind (including
the downstream stages it deletes without recreating, which nothing
would ever re-enqueue). An empty stageIds is a no-op.
Parameters
workflowRunId
string
stageIds
string[]
Returns
Promise<number>
Implementation of
dequeue()
dequeue(
options?):Promise<DequeueResult|null>
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:252
Atomically dequeue the next available job
Parameters
options?
Returns
Promise<DequeueResult | null>
Implementation of
enqueue()
enqueue(
options):Promise<string>
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:177
Add a job to the queue.
Idempotent on (workflowRunId, stageId) per the JobQueue contract:
any row already queued for that pair is removed first, so exactly one
job row exists per stage per run with attempt back at 0 — matching
the @@unique([workflowRunId, stageId]) the reference Prisma schema
declares, so run.rerunFrom and run.reapStuck behave here exactly
as they do against a real database.
Parameters
options
Returns
Promise<string>
enqueueParallel()
enqueueParallel(
jobs):Promise<string[]>
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:218
Enqueue multiple stages in parallel (same execution group).
Idempotent on (workflowRunId, stageId): at most one job row may
exist per stage per run, so an implementation MUST replace any row(s)
already queued for a pair it is asked to enqueue, resetting attempt,
status, workerId, lockedAt, lastError and nextPollAt. That
makes run.rerunFrom's re-enqueue and run.reapStuck's
PENDING-without-job recovery sweep safe to run over a stage that
still carries a terminal job row from a previous execution.
Parameters
jobs
Returns
Promise<string[]>
Implementation of
expireRunawayJobs()
expireRunawayJobs(
absoluteTimeoutMs):Promise<number>
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:553
The coarse tier of the two-tier expiry — see JobQueue.expireRunawayJobs.
Keyed on startedAt, which the heartbeat never refreshes, so it fires on
a worker that is alive but wedged as readily as on one that died.
Parameters
absoluteTimeoutMs
number
Returns
Promise<number>
Implementation of
fail()
fail(
jobId,error,shouldRetry?,fence?):Promise<JobAckOutcome>
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:463
Mark job as failed. shouldRetry defaults to false -- callers must
opt in to a retry rather than risk an unbounded retry loop for
adapters/hosts that omit the argument.
Passing a fence conditions the write on the job still being the RUNNING
attempt with that startedAt; omitting it keeps the previous unconditional
behaviour and always returns "acknowledged". Note that the fenced form is
the recommended one and that the unfenced form exists for transports that
cannot carry the stamp.
Parameters
jobId
string
error
string
shouldRetry?
boolean = false
fence?
Returns
Promise<JobAckOutcome>
Implementation of
getAllJobs()
getAllJobs():
JobRecord[]
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:637
Get all jobs for inspection
Returns
getJob()
getJob(
jobId):JobRecord|null
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:653
Get a specific job by ID
Parameters
jobId
string
Returns
JobRecord | null
getJobsByStatus()
getJobsByStatus(
status):JobRecord[]
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:644
Get jobs by status for inspection
Parameters
status
Returns
getJobsByWorkflowRun()
getJobsByWorkflowRun(
workflowRunId):Promise<JobRecord[]>
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:580
Get all job rows for a workflow run (any status). Used to detect
pending/in-flight retries for a stage (so run.transition doesn't
treat a FAILED stage with a queued retry as terminal) and to find
orphaned SUSPENDED job rows or PENDING stages missing a queued job.
Parameters
workflowRunId
string
Returns
Promise<JobRecord[]>
Implementation of
getWorkerId()
getWorkerId():
string
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:661
Get the worker ID for this queue instance
Returns
string
releaseStaleJobs()
releaseStaleJobs(
staleThresholdMs?):Promise<number>
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:519
Release stale locks (for crashed workers). Stamps lastError with
the LEASE_HEARTBEAT_LOST prefix so an operator can tell a reclaimed
lease from a stage-level failure. The fine-grained tier of a two-tier
expiry whose coarse tier is expireRunawayJobs.
Parameters
staleThresholdMs?
number = 300000
Returns
Promise<number>
Implementation of
resumeJob()
resumeJob(
jobId):void
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:686
Move a suspended job back to pending (for manual resume testing)
Parameters
jobId
string
Returns
void
setDefaultMaxAttempts()
setDefaultMaxAttempts(
maxAttempts):void
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:668
Set max attempts for new jobs
Parameters
maxAttempts
number
Returns
void
setJobLockedAt()
setJobLockedAt(
jobId,lockedAt):void
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:702
Set lockedAt for testing stale job scenarios
Parameters
jobId
string
lockedAt
Date
Returns
void
setJobNextPollAt()
setJobNextPollAt(
jobId,nextPollAt):void
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:728
Set nextPollAt for testing suspended job polling
Parameters
jobId
string
nextPollAt
Date | null
Returns
void
setJobStartedAt()
setJobStartedAt(
jobId,startedAt):void
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:718
Set startedAt for testing the absolute lease cap. Unlike lockedAt,
startedAt is stamped once per claim and no heartbeat refreshes it, so
this is the dial for "a worker that is alive but wedged".
Parameters
jobId
string
startedAt
Date
Returns
void
simulateCrash()
simulateCrash(
jobId):void
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:675
Simulate a worker crash by releasing a job's lock without completing it
Parameters
jobId
string
Returns
void
suspend()
suspend(
jobId,nextPollAt,fence?):Promise<JobAckOutcome>
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:384
Mark job as suspended (for async-batch).
Passing a fence conditions the write on the job still being the RUNNING
attempt with that startedAt; omitting it keeps the previous unconditional
behaviour and always returns "acknowledged". Note that the fenced form is
the recommended one and that the unfenced form exists for transports that
cannot carry the stamp.
Parameters
jobId
string
nextPollAt
Date
fence?
Returns
Promise<JobAckOutcome>
Implementation of
touchJob()
touchJob(
jobId):Promise<void>
Defined in: packages/workflow-engine/src/testing/in-memory-job-queue.ts:589
Refresh a running job's lease (lockedAt) without changing status.
Called periodically by hosts while a long-running stage executes so
releaseStaleJobs doesn't duplicate work still in-flight.
Parameters
jobId
string
Returns
Promise<void>