Queues and jobs

Every side effect that leaves the cluster happens in a BullMQ job. The API writes rows and enqueues; the worker transmits. Redis is used exclusively as the queue backend — there is no application cache in it — so a Redis outage delays dispatch but can never lose or corrupt order data.

Redis is provisioned from the Chrono redis template with HA defaults (three Redis plus three Sentinel replicas). When REDIS_SENTINELS is set, apps/api/src/queues/connection.ts connects through Sentinel for failover and REDIS_URL is ignored; locally it is a single redis:7.2-alpine node on redis://localhost:6379. Every connection sets maxRetriesPerRequest: null, which BullMQ requires: with a finite value a blocking BRPOPLPUSH that outlives the retry budget makes ioredis throw and the worker process dies. Producers share one connection (getSharedRedis()); each Worker and each QueueEvents gets its own from createRedisConnection(label), because a blocking read monopolises a connection.

The queues

QUEUE_NAMES in packages/types/src/queue.ts declares three names, shared so producers and consumers cannot drift. Only two of them have a Queue and a Worker behind them.

Queue name	Produced by	Consumed by	Concurrency	Purpose
supplier-dispatch	API placeOrder fan-out (apps/api/src/domain/orders.ts) and the reroute handler, both via enqueueSupplierDispatch	lucille-worker — processSupplierDispatch	1, hard-coded	Transmit one order to one supplier
backorder-reroute	API inbound-EDI processor (inbound-ack.ts) and the admin backorder/reroute routes, via enqueueBackorderReroute	lucille-worker — processBackorderReroute	EMAIL_WORKER_CONCURRENCY (4)	Move a back-ordered line to its alternate supplier
sysco-inbound-poll	Nothing. No Queue is constructed for this name	Nothing. The poll runs as a plain CronJob process	n/a	Reserved name; see below

sysco-inbound-poll is a name only. syscoPollJobSchema exists in packages/types, but no code calls queue.add() on that queue and no Worker is registered for it. The inbound poll is a Kubernetes CronJob (infra/{dev,prod}/lucille-sysco-poll.yaml, cronSchedule: "*/15 * * * *") that runs node apps/worker/dist/jobs/sysco-poll.js directly. The only place the name has an effect is watchDeduplication() when called with no arguments — as the API server does — which opens a QueueEvents listener for all three names, including the unused one.

supplier-dispatch

One job per supplier per order, not per line item. Eight Sysco lines produce one job carrying eight line items, which becomes one EDI file with eight PO1 segments. The API validates the payload with supplierDispatchJobSchema before enqueuing, so a shape mistake fails on the side that owns the contract.

Payload — SupplierDispatchJob

Field	Type	Required	Notes
orderId	uuid	yes	Part of the deduplication id
dispatchId	uuid	yes	The supplier_dispatches row this job owns
supplierId	uuid	yes
supplierType	sysco | costco | central_kitchen	yes	Selects the transport handler; part of the deduplication id
restaurantId	uuid	yes
restaurantName	string (min 1)	yes	Used in EDI file names, email subjects and PDF headers
restaurantLocation	string | null	no
orderNumber	string (min 1)	yes	Human-facing PO number
placedByEmail	email	no
deliveryNotes	string | null	no	Falls back to orders.notes in the handler
lineItems	array of line items (min 1)	yes	See below

Each entry of lineItems:

Field	Type	Required	Notes
lineItemId	uuid	yes
catalogItemId	uuid	yes
name	string (min 1)	yes	catalog_items.name, for email and PDF rows
supplierSku	string (min 1)	yes	PO107 product id value
supplierProductCode	string | null	no
quantity	positive number	yes
unit	string (min 1)	yes
unitSize	string | null	no

supplierType selects the handler, and each handler returns the reference that is written to supplier_dispatches.reference plus the extra keys appended to the SUPPLIER_SUBMITTED audit event:

supplierType	Handler does	reference is	Extra audit payload keys
sysco	Generate X12 850, allocate control numbers, SFTP put	the EDI file name	remotePath, segmentCount, lineCount, controlNumbers
costco	Render an HTML order table, SMTP send	the SMTP message id	subject, recipients
central_kitchen	Render HTML → PDF via Chromium, upload to Spaces, attach to email, SMTP send	the SMTP message id	pdfUrl, pdfKey, pdfBytes, subject

Every audit payload also carries attempt, reference and supplierType.

Deduplication: Simple Mode, not raw jobId

Both producers go through enqueueSupplierDispatch, which sets nothing but the deduplication id; the retry and retention options come from defaultJobOptions() on the Queue as defaultJobOptions.

import { dispatchDeduplicationId } from '@lucille/types';

// apps/api/src/queues/index.ts — defaults applied once, at Queue construction
export function defaultJobOptions(): JobsOptions {
  const { queue } = getConfig();
  return {
    attempts: queue.attempts, // DISPATCH_ATTEMPTS, default 3
    backoff: { type: 'exponential', delay: queue.backoffMs }, // DISPATCH_BACKOFF_MS, default 1000
    removeOnComplete: { age: 3_600, count: 1_000 },
    removeOnFail: { age: 86_400 },
  };
}

// ...and per enqueue, only the deduplication id:
await supplierDispatchQueue().add('dispatch', payload, {
  deduplication: { id: dispatchDeduplicationId(payload.orderId, payload.supplierType) },
});

BullMQ offers two mechanisms that are frequently confused, and only one of them is correct here.

Custom jobId uniqueness — adding a job whose ID already exists is ignored and emits a duplicated event. The fatal flaw: jobs removed from the queue are no longer considered duplicates. With removeOnComplete: true, a completed job's ID is freed immediately, so a re-enqueue with the same ID succeeds and dispatches to the supplier a second time. That is a duplicate purchase order on Sysco's ERP.

The structured deduplication: { id } option in Simple Mode — the mechanism this code uses, per docs/research/bullmq-idempotency-and-job-deduplication.md §2a. No jobId is ever passed. The deduplication key is kept alive without expiry for as long as the job exists in a non-terminal state, and is released only on terminal completion or terminal failure. Critically, the key survives automatic retries — the code comment in apps/api/src/queues/index.ts states it exactly that way: "the dedup key then survives every automatic retry attempt and is released only when the job reaches a terminal state". A job on attempt 2 of 3 still holds its key, so a concurrent enqueue is still suppressed. Every suppressed enqueue emits a deduplicated event.

The two id formats come from helpers in packages/types/src/queue.ts, so producer and test compute the same string:

Helper	Format
dispatchDeduplicationId(orderId, supplierType)	dispatch:{orderId}:{supplierType}
rerouteDeduplicationId(lineItemId, originalSupplierId)	reroute:{lineItemId}:{originalSupplierId}

enqueueSupplierDispatch takes an optional { dedupSuffix }, which appends :{suffix} to the dispatch id. It exists for exactly one caller: the reroute handler, which passes dedupSuffix: 'reroute:{lineItemId}'. Without it, a reroute to an alternate supplier that happens to be the same type as the original (Costco → Costco, say) would compute dispatch:{orderId}:{supplierType} — colliding with the order's original dispatch — and the reroute would be silently swallowed as a duplicate.

// apps/worker/src/handlers/backorder-reroute.ts
const jobId = await enqueueSupplierDispatch(dispatchPayload, {
  dedupSuffix: `reroute:${lineItem.id}`,
});

Two behaviours to keep in mind:

• The key is released on failure. Once all retries are exhausted, a re-enqueue with the same key succeeds and creates a new job. That is correct for a legitimate manual retry, and it is exactly why the worker also needs its own idempotency check.
• job.remove() clears the key. Never call it on an active dispatch job from admin tooling without re-checking the dispatch's database state first.

debounceId and debounce: { id } are deprecated; use deduplication: { id } in all new code.

Worker-level idempotency: the second line of defence

Queue-level deduplication prevents duplicate enqueues. It does not prevent a job from executing twice. BullMQ's stall detection re-queues a job that was mid-execution when its worker crashed, and that re-queue bypasses deduplication entirely.

So processSupplierDispatch reads PostgreSQL before touching the outside world. The check is precise: it loads the supplier_dispatches row by primary key = job.data.dispatchId and inspects the single column supplier_dispatches.status (enum supplier_dispatch_status: pending, submitted, confirmed, failed). If it is submitted or confirmed, the handler returns without transmitting.

const dispatch = await prisma.supplierDispatch.findUnique({
  where: { id: payload.dispatchId },
  include: DISPATCH_INCLUDE,
});

if (!dispatch) {
  job.discard?.();
  throw new PermanentDispatchError(
    `supplier_dispatches row ${payload.dispatchId} does not exist — nothing to dispatch`,
  );
}

if (dispatch.status === 'submitted' || dispatch.status === 'confirmed') {
  return { dispatchId: dispatch.id, skipped: true, reason: 'already_submitted' };
}

An SFTP drop and an email are not reversible, so this row — not the BullMQ job — is the authoritative record of whether a supplier has been contacted. On success markSubmitted sets the dispatch to submitted with its reference, submittedAt and attempt, and its line items to submitted, in one prisma.$transaction; the SUPPLIER_SUBMITTED audit event and the order status roll-up are written afterwards, deliberately outside the transaction. Note there is no NonRetryableError class in this codebase: the permanent-failure type is PermanentDispatchError, and BullMQ is stopped from retrying by job.discard().

The enqueue atomicity gap

There is one residual failure mode noted in the research brief. If the process crashes between the Prisma transaction committing and queue.add() returning, the order exists and no job was ever scheduled.

What the code does today is narrower than a full outbox: placeOrder wraps each enqueueSupplierDispatch in a try/catch and, on an enqueue error, sets that dispatch to failed with Could not enqueue dispatch: … and appends SUPPLIER_DISPATCH_FAILED. That covers a failed enqueue but not a crash mid-enqueue. A transactional outbox and a reconciliation sweep over stale pending dispatches are designed but not implemented — no relay job, repeatable job or sweeper for this exists in the repository.

Retry and backoff

Setting	Value	Set by
attempts	3	DISPATCH_ATTEMPTS (default 3)
backoff.type	exponential	hard-coded
backoff.delay	1000 ms	DISPATCH_BACKOFF_MS (default 1000)
Retry delays	BullMQ computes delay × 2^(attemptsMade−1), so with 3 attempts the two retries wait 1 s then 2 s	BullMQ 5.81.x
On final failure	supplier_dispatches.status = 'failed' (+ attempt, failureReason), line items dispatch_failed, order recomputed to partial or failed, SUPPLIER_DISPATCH_FAILED audit event, admin alert email, Sentry event	markFailed + sendAdminAlert

The alert email goes to ADMIN_ALERT_EMAIL; sendAdminAlert is a no-op when that variable is unset.

Transient and permanent errors are treated differently. isPermanentFailure() classifies PermanentDispatchError, SupplierConfigError, PermanentMailError, SftpConfigError, ZodError, ChromiumUnavailableError and anything carrying permanent === true as permanent; everything else is retryable, on the stated grounds that over-classifying as permanent loses an order a second attempt would have delivered. A permanent failure calls job.discard() — BullMQ's documented "do not retry this even though attempts allows it" flag — and fails the dispatch immediately rather than burning three attempts. ssh2-sftp-client has no built-in connection retry, so BullMQ is the sole retry mechanism for Sysco.

Job retention — why not true

removeOnComplete: { age: 3_600, count: 1_000 },   // NOT true
removeOnFail:     { age: 86_400 },                // NOT true

Three reasons the boolean is wrong here:

1. Failed jobs are the audit trail of a bad afternoon. Keeping them 24 hours means an engineer can read the failure and retry it by hand.
2. removeOnComplete: true frees the job ID immediately, which breaks jobId-based deduplication for any re-enqueue inside the same business window. Less critical with the deduplication API, since the key is a separate record, but there is no upside to aggressive removal.
3. Auto-removal is lazy. BullMQ prunes the completed set only when a job completes, and the failed set only when a job fails. Lowering a count limit does not take effect until the next event of that kind, which on a low-volume queue can be a long time. The research brief suggests a queue.clean() on startup as a mitigation; the worker does not currently do that.

Concurrency

apps/worker/src/worker.ts registers the supplier-dispatch worker with concurrency: 1 hard-coded, for every supplier type — not only for Sysco. The reason is Sysco's SSH session limit: concurrent connections trip it with "no more sessions", so the Sysco lane must be serialised, and BullMQ offers no server-side filter that would let one shared queue run two lanes at different concurrencies. The rejected alternatives are documented in the file's header comment; the stated fix, if throughput ever matters, is a dedicated Sysco queue chosen by the producer. The trade-off is explicit: Costco emails and Central Kitchen PDF renders are serialised too, which at roughly 25 dispatches a day is acceptable.

The dispatch worker also sets lockDuration: 120_000 so a Chromium render plus an S3 upload plus an SMTP conversation cannot be mistaken for a stalled job.

backorder-reroute only touches Postgres and Redis, so it runs at EMAIL_WORKER_CONCURRENCY (default 4).

Deviation from the plan: SYSCO_WORKER_CONCURRENCY (default 1) is still parsed by apps/api/src/config/index.ts and exposed as config.queue.syscoConcurrency, but no code reads it. Setting it has no effect today; the value it would control is the literal 1 in worker.ts.

Queue-related environment variables

Variable	Default	Effect
QUEUE_PREFIX	lucille	BullMQ key prefix — not the BullMQ default bull
DISPATCH_ATTEMPTS	3	attempts on every job
DISPATCH_BACKOFF_MS	1000	Exponential backoff base delay
SYSCO_WORKER_CONCURRENCY	1	Parsed, currently unused (see above)
EMAIL_WORKER_CONCURRENCY	4	Concurrency of the backorder-reroute worker
REDIS_URL	redis://localhost:6379	Ignored when REDIS_SENTINELS is set
REDIS_SENTINELS	(unset)	Comma-separated host:port list; switches to Sentinel mode
REDIS_SENTINEL_NAME	mymaster	Sentinel master name
WORKER_PORT	3001	Worker health/metrics HTTP port

Observability

watchDeduplication() attaches a QueueEvents listener per queue. Its deduplicated handler increments the Prometheus counter queue_jobs_deduplicated_total (label: queue) and logs at warn with { queue, jobId, deduplicationId }. It does not report to Sentry — only the worker's failed and error listeners call captureError. A trickle of deduplications is normal, because chefs double-click; a spike means a producer is re-enqueuing something it believes was lost.

Worker lifecycle events feed queue_jobs_total (labels queue, outcome = completed/failed), and the dispatch handler records supplier_dispatch_duration_seconds and supplier_dispatch_failures_total. The reroute handler records backorder_reroutes_total{outcome}. See observability.

Graceful shutdown

SIGTERM and SIGINT call stopWorker, which calls worker.close() without force on each worker so in-flight jobs finish — a half-written EDI file on Sysco's server is worse than a slow rollout — then closes the health server, the queues, Redis, Prisma and telemetry, and flushes Sentry. A second signal during drain is logged and ignored. The worker also serves /healthz (liveness, no external calls), /readyz (Postgres, Redis and Chromium) and /metrics from apps/worker/src/health.ts.

backorder-reroute

Payload — BackorderRerouteJob

Field	Type	Required	Notes
orderId	uuid	yes
lineItemId	uuid	yes	Part of the deduplication id
originalSupplierId	uuid	yes	The supplier that back-ordered; part of the deduplication id
alternateSupplierId	uuid | null	no	Omitted rather than null when unknown; the handler resolves it
reason	string (max 500)	no	Free text, e.g. Supplier acknowledgment IB on <file>
ackCode	IA IB IR IC IP IQ DR BP SP	no	The X12 855 ACK01 code that triggered the reroute

There is no trigger or quantity field. An automatic reroute is distinguished from an admin-initiated one by which producer enqueued it and by the actorId on the preceding LINE_ITEM_BACKORDERED audit event, not by a payload flag.

Handler

1. Load the line item with its order, catalog item, supplier mapping and every dispatch it has been part of. A missing line item throws — that is worth retrying.
2. Already rerouted → a linked dispatch with is_reroute = true exists, so return outcome: 'discarded', reason: 'already_rerouted'.
3. Already reported as a dead end → an append-only BACKORDER_NO_ALTERNATE audit event for this (lineItemId, originalSupplierId) exists, so return reason: 'already_no_alternate' rather than emailing the administrator twice.
4. No usable alternate — none mapped, or the candidate is missing, inactive, or is the supplier that back-ordered the line → set the line to backorder_no_alternate, append BACKORDER_NO_ALTERNATE, increment backorder_reroutes_total{outcome="no_alternate"}, email the administrator, and complete successfully. "No alternate" is a business outcome, not a job failure: throwing would burn three attempts, alert three times and pollute the failed set.
5. Alternate configured → in one transaction create a new supplier_dispatches row (is_reroute = true, reroute_of_dispatch_id pointing at the original), link the line item, and set the line to rerouted. Then append BACKORDER_REROUTED and enqueue a supplier-dispatch job for the alternate with the dedupSuffix described above. The original dispatch is never mutated — it is the record of what Sysco was actually sent.

Idempotency

Queue-level: reroute:{lineItemId}:{originalSupplierId}. Sysco can legitimately send the same acknowledgment twice — a file re-dropped before the previous poll archived it — and a duplicate reroute would place a second order with the alternate supplier. Database-level: steps 2 and 3 above, which is what covers a stalled-and-requeued job after the dedup key has been released.

sysco-inbound-poll (CronJob)

Not a queue: a Kubernetes CronJob on */15 * * * * running node apps/worker/dist/jobs/sysco-poll.js from the shared image, with concurrencyPolicy: Forbid fixed by the chart.

for each active supplier of type sysco:
  withSftp(...)                       # one connection per supplier per run, end()ed unconditionally
    list inboundDir
    skip names starting '.' or ending '.tmp' / '.part'
    for each candidate (for...of with await — never .forEach):
        getFile
        processInboundEdi(content, { supplierId, reference: filename, parsers })
        parsed        -> move to archiveDir
        not parsed    -> ensureDir + move to {archiveDir}/failed, alert the admin
        threw         -> leave the file in place, Sentry, count as failed
finally: close queues, Redis, Prisma, telemetry; flush Sentry; exit explicitly

The poll does not POST to /api/inbound/sysco. It calls processInboundEdi in-process, because there is no machine-to-machine Zitadel credential and the CronJob already runs from the same image with the same database credentials; that route remains the admin-driven reprocess entrypoint. The reroute jobs are therefore enqueued by processInboundEdi inside this same process.

Failure semantics are deliberately forgiving. Archiving only after successful processing is what makes the poll idempotent at the file level, and processInboundEdi additionally keys on vendor_documents.reference so a re-read reports alreadyProcessed. The exit code is 1 on an SFTP connection or listing failure, and 1 when a non-empty directory yielded nothing processed or skipped (a systemic problem); a single bad file among good ones exits 0, since it is already in Sentry and the admin mailbox.

Local inspection

The redis Chrono template includes a bull-monitor dashboard for queue visibility in the cluster. Locally, point any BullMQ UI at redis://localhost:6379. When inspecting keys directly, remember the prefix is QUEUE_PREFIX (lucille), not BullMQ's default bull, and that BullMQ's deduplication key is {prefix}:{queue}:de:{deduplicationId}:

# Which queues exist and how deep are they?
docker compose exec redis redis-cli --scan --pattern 'lucille:*:meta'
docker compose exec redis redis-cli llen lucille:supplier-dispatch:wait

# Live deduplication keys, e.g. lucille:supplier-dispatch:de:dispatch:<orderId>:sysco
docker compose exec redis redis-cli --scan --pattern 'lucille:supplier-dispatch:de:*'

Verification status

This page was checked line by line against apps/api/src/queues/index.ts, apps/api/src/queues/connection.ts, apps/api/src/config/index.ts, packages/types/src/queue.ts, apps/worker/src/worker.ts, apps/worker/src/handlers/{supplier-dispatch,backorder-reroute,sysco,costco,central-kitchen}.ts, apps/worker/src/jobs/sysco-poll.ts, apps/worker/src/health.ts, apps/api/prisma/schema.prisma, infra/{dev,prod}/lucille-sysco-poll.yaml and BullMQ 5.81.3 itself (for the backoff formula and the de: key layout).

Automated test coverage is uneven, and it is worth being blunt about where.

Covered — the producers and the contract. packages/types/src/__tests__/queue.test.ts asserts the three queue names, both deduplication id formats and the payload schemas. apps/api/tests/orders.place.test.ts, orders.backorder.test.ts and orders.idempotency.test.ts assert against a real local Redis that the right jobs with the right payloads land on the right queues, with no BullMQ mock.

Covered — the dispatch handler, for Sysco. apps/worker/tests/handlers/sysco.test.ts (4 tests, passing) drives processSupplierDispatch end to end with nothing mocked: a seeded order in lucille_test, a real in-process SFTP server, the real X12 850 serialiser and real control-number allocation. It asserts the generated file name pattern and that it is the only file in the outbound tree; the parsed envelope (ISA sender/receiver, T usage indicator, one PO group, 850, SE01 equal to the real ST-through-SE count, matching trailer control number); BEG03 equal to the order number; one PO1 per line with the right quantities, UOM mapping and SKUs; CTT01; the N1 ship-to naming the restaurant; then the database side — supplier_dispatches.status = 'submitted' with reference, submittedAt, attempt = 1, null failureReason and null pdfUrl, line items submitted, the order rolled up to completed — and the audit trail, including the SUPPLIER_SUBMITTED payload's reference, supplierId, supplierType, lineCount, segmentCount and controlNumbers. Two further cases prove ISA13 is allocated fresh and monotonically per interchange, that a missing outbound directory is created, and that the SFTP session is opened once per job and left closed. So the handler's happy path, its state transitions and its audit trail are genuinely verified; the early-return skip on a dispatch that is already submitted or confirmed is not directly exercised by any test — only the skipped: false path is.

Present but failing. apps/worker/tests/handlers/costco.test.ts (2 tests) covers the SMTP path against a real Mailpit relay — subject format, To/Cc/From, every line item in both the HTML and plain-text alternatives, exactly one message sent, the message id stored as the dispatch reference — and, in its second case, the permanent-failure path: a Costco supplier with no emailAddress rejects, job.discard() is called, and the dispatch is failed on attempt 1 of 3. Both cases were failing at the time of writing (two assertion mismatches, reported as being around the expected recipient address and the list of messages read back from Mailpit), in a newly added file. Treat the retry-classification and email-content claims above as not-yet-passing coverage rather than verified.

Not covered. There is still no test file for apps/worker/src/worker.ts or for apps/worker/src/jobs/sysco-poll.ts, and none for handlers/backorder-reroute.ts or handlers/central-kitchen.ts. So the concurrency and lockDuration settings, the deduplicated listener, graceful shutdown, the reroute decision tree and the whole inbound poll cycle are verified by reading the code only. apps/worker/tests/helpers/queues.ts exists and is written for a reroute-handler test that has not been added yet. The rest of apps/worker/tests/ — edi/, email/, pdf/, sftp/ plus fixtures/ and helpers/ — tests the X12 writer, the 855/997 parsers, the email templates, the PDF template and renderer and the SFTP adapter in isolation.

Designed but not implemented. The transactional outbox, the reconciliation sweep over stale pending dispatches, queue.clean() on worker startup, and any use of SYSCO_WORKER_CONCURRENCY.

The worker runtime was under active development while this page was written, so if a detail here looks off, re-check apps/worker/src/ and apps/worker/tests/ before trusting the page.

Where to go next

• Supplier integrations — what each handler actually does.
• Order lifecycle — how job outcomes become statuses.
• Observability — the metrics and log fields named above.
• Testing — the harnesses that exist and how they are run.
• Troubleshooting — the failure modes these settings guard against.