Interface: JobTransport
Defined in: packages/workflow-engine/src/kernel/ports.ts:272
Job queue abstraction.
Same shape as the existing JobQueue interface so that current
implementations (InMemoryJobQueue, PrismaJobQueue) structurally satisfy
this port without adapters.
Properties
fairnessGroupBy?
readonlyoptionalfairnessGroupBy?:string|null
Defined in: packages/workflow-engine/src/kernel/ports.ts:279
The dotted groupBy path this transport's fairness cap reads, or null
when fairness is off (or undefined when unsupported). Read by
createSpillingJobTransport so a spilled payload still carries its group
key.
Methods
adoptWorkerId()?
optionaladoptWorkerId(workerId):string
Defined in: packages/workflow-engine/src/kernel/ports.ts:418
Optional. Offer the host's workerId to the transport, so the id
stamped on job_queue.workerId is the one that identifies the
process — transports are usually constructed before the host and
would otherwise invent their own (worker-<pid>-<ts>), leaving
"which worker ran this stage" unanswerable from the job row.
A transport that was explicitly configured with a worker id MUST keep it (the caller said what it wanted). Either way it returns the id it will actually stamp, so the host can warn when the two disagree.
Parameters
workerId
string
Returns
string
cancelByRun()
cancelByRun(
workflowRunId):Promise<number>
Defined in: packages/workflow-engine/src/kernel/ports.ts:396
Cancel all pending/suspended jobs for a workflow run. Returns count cancelled.
Parameters
workflowRunId
string
Returns
Promise<number>
complete()
complete(
jobId,fence?):Promise<JobAckOutcome>
Defined in: packages/workflow-engine/src/kernel/ports.ts:319
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>
defer()?
optionaldefer(jobId,nextPollAt,reason,fence?):Promise<JobAckOutcome>
Defined in: packages/workflow-engine/src/kernel/ports.ts:348
Return a claimed job to PENDING with a later nextPollAt WITHOUT counting
the claim as an attempt (the dequeue incremented attempt; this undoes it).
fail(id, err, true) is the wrong shape for work a host declines rather
than fails: declining is not a failed attempt, and a condition that lasts
for a whole deploy — a run pinned to a version this build does not
present — exhausts the three-attempt budget in about fifteen seconds and
takes the job row terminal.
Optional: a transport that does not implement it falls back to fail with retry, which is correct but bounded by the budget.
Parameters
jobId
string
nextPollAt
Date
reason
string
fence?
Returns
Promise<JobAckOutcome>
deleteByRunAndStages()
deleteByRunAndStages(
workflowRunId,stageIds):Promise<number>
Defined in: packages/workflow-engine/src/kernel/ports.ts:302
Remove every job row for the given stages of a run, whatever their
status; returns how many were removed. run.rerunFrom calls it for
the stage records it deletes. An empty stageIds is a no-op.
Parameters
workflowRunId
string
stageIds
string[]
Returns
Promise<number>
dequeue()
dequeue(
options?):Promise<DequeueResult|null>
Defined in: packages/workflow-engine/src/kernel/ports.ts:308
Atomically dequeue the next available job.
Parameters
options?
Returns
Promise<DequeueResult | null>
enqueueParallel()
enqueueParallel(
jobs):Promise<string[]>
Defined in: packages/workflow-engine/src/kernel/ports.ts:295
Enqueue multiple stages in parallel (same execution group).
Idempotent on (workflowRunId, stageId): at most one job row exists
per stage per run, so a transport MUST replace any row already queued
for a pair it is asked to enqueue, resetting attempt, status,
workerId, lockedAt, lastError and nextPollAt. run.rerunFrom
and run.reapStuck's PENDING-without-job sweep both re-enqueue a
stage that may still carry a terminal job row from a previous
execution; a transport that inserts unconditionally either
accumulates duplicate rows or (on a schema declaring the
@@unique([workflowRunId, stageId]) the reference schema ships)
fails the insert.
Parameters
jobs
Returns
Promise<string[]>
expireRunawayJobs()?
optionalexpireRunawayJobs(absoluteTimeoutMs):Promise<number>
Defined in: packages/workflow-engine/src/kernel/ports.ts:393
Fail every RUNNING job whose claim (startedAt, stamped once and never
refreshed) is older than absoluteTimeoutMs, stamping lastError with
the LEASE_ABSOLUTE_CAP prefix; returns how many. The coarse tier of a
two-tier expiry: releaseStaleJobs is the fine-grained heartbeat
signal and is defeated by a worker that is alive but wedged, because
such a worker keeps calling touchJob. Optional — a transport that
does not implement it simply has no absolute cap.
Parameters
absoluteTimeoutMs
number
Returns
Promise<number>
fail()
fail(
jobId,error,shouldRetry?,fence?):Promise<JobAckOutcome>
Defined in: packages/workflow-engine/src/kernel/ports.ts:369
Mark job as failed. With shouldRetry: true the transport MUST put the
job back in the queue (PENDING, with backoff, keeping its attempt
count) — the kernel has already recorded the stage as PENDING on that
promise, and a transport that only acknowledges the message leaves the
run RUNNING until run.reapStuck heals it. With false the job is
terminal and the host dispatches run.transition right away.
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
fence?
Returns
Promise<JobAckOutcome>
getJobsByWorkflowRun()
getJobsByWorkflowRun(
workflowRunId):Promise<JobRecord[]>
Defined in: packages/workflow-engine/src/kernel/ports.ts:402
Get all job rows for a workflow run (any status). Used to detect pending/in-flight retries for a stage and to find orphaned job rows.
Parameters
workflowRunId
string
Returns
Promise<JobRecord[]>
releaseStaleJobs()
releaseStaleJobs(
staleThresholdMs?):Promise<number>
Defined in: packages/workflow-engine/src/kernel/ports.ts:382
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
Returns
Promise<number>
suspend()
suspend(
jobId,nextPollAt,fence?):Promise<JobAckOutcome>
Defined in: packages/workflow-engine/src/kernel/ports.ts:330
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>
touchJob()
touchJob(
jobId):Promise<void>
Defined in: packages/workflow-engine/src/kernel/ports.ts:405
Refresh a running job's lease without changing status.
Parameters
jobId
string
Returns
Promise<void>