04 · Jobs engine (jobs, lanes, schedules, Kafka target)

As-built reference for the TMS v2 phase-1 jobs engine. Repo for every path:line below is bigc-tms-core-v2 (worktree wt-v2-prototype-core) unless a line says store-v2.

1. Summary

The jobs engine is core-v2's one durable work queue. A module calls sideEffect.enqueue(target, payload, options, tx) inside its own Mongo transaction, so a committed business change always has its job rows and a rolled-back one has none. Rows go to one of two lanes. The fast lane is for short sends such as Kafka: 64 partitions, one owner pod per partition, rows run in key + version order. The long lane is for handlers that take minutes: up to 10 rows are reserved with a claim token and a 5-min lease that is renewed every minute. When the target answered with a failure, the row is retried 5 times (1/2/4/8/16 min) and then goes DEAD. When the target could not be reached, the target is paused and no attempt is spent. Schedules are crons declared in code: every pod ticks, and each slot becomes one long-lane job whose dedupe key is <code>:<slot>. Admins list jobs, retry or skip DEAD rows, and enable, disable or override schedules.

repo	branch	sha	folders
bigc-tms-core-v2	origin/main-v2-prototype	3d70475a (jobs feature 38dad930; changed by 3dd1c5a0: Kafka target moved to src/kafka, job_stats and demo schedules removed, request context added)	src/jobs/, src/schedules/, src/kafka/kafka.target.ts, src/common/utils/transaction.ts, caller src/delivery_order/delivery_order.service.ts
bigc-tms-store-v2	main-v2-prototype	0851a3a	src/pages/setting/outbound-events/, src/pages/api/outbound_events/, src/apis/{client,server}/outbound_event/

2. File map

file	responsibility
src/jobs/jobs.constants.ts	Enums JobStatus (PENDING/RUNNING/DEAD), JobLane, DeliveryMode, and every tuning number: partitions, batch sizes, retry delays, poll, lease, probe, the JOBS_WORKERS switch
src/jobs/jobs.schema.ts	jobs collection and its 4 indexes
src/jobs/job_partitions.schema.ts	job_partitions (64 lease rows) and job_workers (one liveness row per pod)
src/jobs/job_targets.registry.ts	In-memory target registry: name → lane, handler, lossy, delivery mode, exclusive; per-code max attempts
src/jobs/side_effect.service.ts	Enqueue API (enqueue, enqueueMany), partition hash, dedupe pre-filter, lossy hand-off to Bull after commit
src/common/utils/transaction.ts	inTransaction(conn, work): one Mongo transaction plus afterCommit hooks
src/jobs/fast_lane.worker.ts	JobOutcome (shared outcome path for both lanes), FastLaneWorker (poll, ordered batch, per-row in-flight mark, takeover close), jobContext, nextPollDelay, retryDelayMs
src/jobs/partition_lease.service.ts	Partition seeding and repair, lease heartbeat, fair-share acquire and release, in-flight marks, pod id
src/jobs/long_lane.worker.ts	Long lane reserve with claim token, lease heartbeat, reclaim after expiry, exclusive-code filter, background batches
src/jobs/target_pause.ts	TargetUnreachable, TargetOutcomeUnknown errors and the per-pod pause map with a 30 s probe
src/jobs/lossy.processor.ts	Bull processor for lossy targets (one try, dropped on failure)
src/jobs/jobs.controller.ts, jobs.dto.ts	Admin API GET /jobs, POST /jobs/:id/retry, POST /jobs/:id/skip
src/jobs/jobs.module.ts	Wires the models, Bull queue, the clock, pod id and the workers; exports JobTargetsRegistry and SideEffectService
src/schedules/schedules.schema.ts	schedules collection
src/schedules/schedules.constants.ts	Missed-slot policy enum, the schedule: target prefix, tick, grace, timezone offset, default retry count
src/schedules/cron.ts	Own 5-field cron parser (next, intervalMs) at a fixed +07:00
src/schedules/schedules.service.ts	declare(), boot sync, tick, slot run (late-slot skip and run_once collapse), admin operations
src/schedules/schedules.controller.ts, schedules.dto.ts	Admin API /schedules
src/schedules/schedules.module.ts	Wires the model, imports JobsModule, exports SchedulesService
src/kafka/kafka.target.ts	Registers the kafka:<topic> fast-lane targets, sends a job through the pod producer, sorts Kafka errors into outcome classes
src/kafka/kafka.producer.ts	One long-lived producer per pod; kafkajs retries are off, connect timeout 3 s, request timeout 10 s
src/delivery_order/delivery_order.service.ts	The only business caller in this build: it enqueues the DO-update Kafka jobs inside the DO transaction

3. Data model

jobs (model job, jobs.schema.ts:14-82)

field	type	req	default	enum / validation	index	note
code	string	yes	—	—	—	Business name of the job. Defaults to target at enqueue (side_effect.service.ts:65). Selects the per-code max attempts (job_targets.registry.ts:49-51)
target	string	yes	—	must be registered when enqueued (side_effect.service.ts:52)	—	kafka:<topic> or schedule:<code>
lane	string	yes	—	fast | long (set from the target, side_effect.service.ts:67)	part of the PENDING index	—
key	string	no	—	—	—	Ordering key, for example the DO number. Doubles as the Kafka message key
partition	number	yes	—	0..63 = md5(key || dedupe_key) mod 64 (side_effect.service.ts:29-32,69)	part of the PENDING index	—
version	number	no	0	—	—	Order inside a key (fast_lane.worker.ts:185)
payload	Mixed	no	—	—	—	Left out of list responses (jobs.controller.ts:20)
dedupe_key	string	yes	uuid v4 when not given (side_effect.service.ts:63)	—	unique	Idempotency key
status	string	no	PENDING	PENDING / RUNNING / DEAD (schema does not enforce the enum)	partial filters	Only the long lane writes RUNNING
attempts	number	no	0	—	—	+1 per answered failure only
next_run_at	Date	no	set to now at enqueue	—	part of the PENDING index	—
last_error	string	no	—	cut to 2000 chars (fast_lane.worker.ts:37)	—	—
claimed_by	string	no	—	pod id	—	long lane
lease_until	Date	no	—	—	partial index while RUNNING	long lane row lease
claim_token	string	no	—	uuid per reserve	—	long lane
outcome_token	string	no	—	—	—	fast lane: the in-flight token under which the outcome was written (jobs.schema.ts:60-62)
running_code	string	no	—	—	unique partial while RUNNING	Set only for exclusive targets (long_lane.worker.ts:95)
closed_at	Date	no	—	—	—	Set when the row goes DEAD
trace_id	string	no	from the request context, or a new one	—	—	side_effect.service.ts:57
actor	Object	no	from the request context	—	—	Opened again as via: system when the job runs (fast_lane.worker.ts:42-47)
headers	Object	no	—	—	—	Extra Kafka headers (kafka.target.ts:83)
created_at, updated_at	Date	auto	—	—	—	timestamps

Indexes (jobs.schema.ts:84-96):

keys	options	serves
{lane, partition, next_run_at}	partial status: PENDING	fast lane read, long lane PENDING read
{lease_until}	partial status: RUNNING	long lane reclaim
{dedupe_key}	unique	enqueue idempotency
{running_code}	unique, partial status: RUNNING, running_code exists	at most one RUNNING row per exclusive code across pods

The collection has no TTL index. A finished row is deleted (fast_lane.worker.ts:93). A DEAD row stays until an admin retries or skips it.

job_partitions (model job_partition, job_partitions.schema.ts:11-32)

field	type	req	default	index	note
partition	number	yes	—	unique (built by hand before seeding, partition_lease.service.ts:42)	0..63
owner	string	no	null	{owner}	pod id
lease_until	Date	no	—	—	now + 30 s, renewed every 10 s
inflight	[string]	no	[]	—	id of the job row being sent now (at most one per row in practice, fast_lane.worker.ts:208)
inflight_token	string	no	—	—	uuid per send. Compared with jobs.outcome_token on takeover

job_workers (model job_worker, job_partitions.schema.ts:42-53)

field	type	req	index	note
pod_id	string	yes	unique	hostname:pid:uuid8 (partition_lease.service.ts:11)
heartbeat_at	Date	yes	{heartbeat_at}	A pod counts as live when this is newer than 30 s. Deleted on clean shutdown (partition_lease.service.ts:104)

schedules (model schedule, schedules.schema.ts:12-44)

field	type	req	default	enum	index	owner
code	string	yes	—	—	unique	code
cron	string	yes	—	5-field cron	—	code (overwritten at every boot, schedules.service.ts:88)
enabled	boolean	no	true	—	part of the partial index	admin (set only on insert, schedules.service.ts:92)
cron_override	string	no	—	5-field cron	—	admin
missed_slot_policy	string	no	run_once	run_once | skip	—	code. Default: skip when the interval is 5 min or less, else run_once (schedules.service.ts:36-38)
retry_count	number	no	3	—	—	code. Used as the total number of runs per slot (schedules.service.ts:99)
last_run_at	Date	no	—	—	—	$max of slot times that ran (schedules.service.ts:184)
last_skipped_at	Date	no	—	—	—	$max of slot times skipped at run time (schedules.service.ts:179)
next_run_at	Date	no	—	—	{next_run_at} partial enabled: true	engine

Redis

key pattern	value	TTL	note
bull:side_effect_lossy:* (Bull defaults on QUEUE_REDIS_URL, app.module.ts:40-42)	lossy job {target, code, key, payload, trace_id, actor, headers}	removed on complete or fail (side_effect.service.ts:119)	Only lossy targets use it, and no lossy target is registered in this build

The engine keeps no other Redis state. Pause state lives in pod memory (target_pause.ts:26-30).

Relations

• jobs.partition → job_partitions.partition (logical). job_partitions.inflight[] → jobs._id.
• jobs.target = schedule:<code> and jobs.code → schedules.code. A slot is jobs.payload.slot (ISO string) with dedupe_key = <code>:<slotISO> (schedules.service.ts:34).
• DO-update jobs: jobs.key = delivery_order.delivery_no, jobs.version = job_counters.seq (delivery_order.service.ts:159-166).

4. API

Global prefix /api, URI versioning v1 (main.ts:32-37), so the routes are /api/v1/.... A ValidationPipe with transform and implicit conversion runs on every route (main.ts:46-49). All routes below need role admin through the global AuthGuard (jobs.controller.ts:22, schedules.controller.ts:15, guard common/guards/auth.guard.ts:34,79-80). Any other role, system_admin included, gets 403. Errors use throwError, which returns {statusCode, code, message[]} with status 400 when none is given (common/utils/error.ts:4-16).

method + route	request	response	errors	code
GET /api/v1/jobs	query status (enum), target, code, key, page ≥ 1, limit 1..100 (default 100)	{data: Job[] (no payload, _id kept, lean), total, page, limit}, sorted _id desc	400 when validation fails	jobs.controller.ts:33-43, jobs.dto.ts:7-36
POST /api/v1/jobs/:id/retry	—	The updated row without payload: status PENDING, attempts 0, next_run_at now; claim fields unset	404 not_found (unknown or invalid id) · 409 invalid_status (not DEAD, or lost a race)	jobs.controller.ts:46-60,71-76
POST /api/v1/jobs/:id/skip	—	{_id, skipped: true}; the row is deleted and a log line written with result=skipped	404 · 409 as above	jobs.controller.ts:62-69
GET /api/v1/schedules	—	every schedules row sorted by code	—	schedules.controller.ts:23-26, schedules.service.ts:116-118
POST /api/v1/schedules/:code/enable	—	the updated row. Going from disabled to enabled resets next_run_at to the next slot after now	404 not_found	schedules.service.ts:120-125
POST /api/v1/schedules/:code/disable	—	the updated row	404	same
PUT /api/v1/schedules/:code/cron_override	body {cron_override: string}	the updated row; next_run_at follows the override	404 · 400 invalid_cron	schedules.service.ts:127-131, schedules.dto.ts:3-6
DELETE /api/v1/schedules/:code/cron_override	—	the updated row; override unset, next_run_at back on the code cron	404	schedules.service.ts:133-137

In-process API (the main one):

call	contract	code
SideEffectService.enqueue(target, payload, {code?, key?, version?, dedupe_key?, headers?}, tx?)	Returns {inserted, lossy}. An unknown target throws unknown_target before any write	side_effect.service.ts:16-27,46-48
SideEffectService.enqueueMany(items, tx?)	One insertMany for every must-not-lose row, joined to tx.session. Lossy rows go to Bull through tx.afterCommit	side_effect.service.ts:51-89
JobTargetsRegistry.register({name, lane, handler, lossy?, delivery_mode?, exclusive?})	A duplicate name throws duplicate_target. Delivery mode defaults to at_most_once	job_targets.registry.ts:7-29
SchedulesService.declare({code, cron, handler, missed_slot_policy?, retry_count?})	Call it from onModuleInit. It validates the cron, throws duplicate_schedule on a repeat code, and registers target schedule:<code> as long lane and exclusive	schedules.service.ts:26-32,55-65
inTransaction(conn, work)	Runs work(tx) in session.withTransaction. Hooks run only after the commit; a hook error is logged, not thrown	common/utils/transaction.ts:12-32
handler errors	TargetUnreachable → pause, no attempt spent · TargetOutcomeUnknown → DEAD at once on at-most-once (pauses unless target_up), treated as unreachable on at-least-once · {dead_now: true} → DEAD at once, pause state left as is · any other error → answered failure	target_pause.ts:5-24, fast_lane.worker.ts:67-83

5. Conditions

when …	the system must …	enforced at	test
a business transaction enqueues and then aborts	leave 0 job rows and 0 Bull jobs	side_effect.service.ts:85,101-103, transaction.ts:17-24	side_effect.service.spec.ts:88,104; delivery_order.service.spec.ts:235
the same dedupe_key is enqueued again (same or a later transaction)	insert nothing and raise no error. The row is filtered out before the insert, because a duplicate-key error would abort the caller's transaction	side_effect.service.ts:94-100, outside a transaction :105-111	side_effect.service.spec.ts:72,82
two items in one call share a dedupe_key	keep the first	side_effect.service.ts:81	untested
the target is not registered	throw before any write	side_effect.service.ts:52, job_targets.registry.ts:35-39	side_effect.service.spec.ts:121
key is absent	take the partition from dedupe_key	side_effect.service.ts:69	side_effect.service.spec.ts:115
the target answered with a failure	add 1 to attempts and wait 1, 2, 4, 8, 16 min. The 6th failure (or the per-code max) → DEAD with closed_at and an job dead alert error log	fast_lane.worker.ts:104-121, jobs.constants.ts:22-23	fast_lane.worker.spec.ts:85,103,130
the target is unreachable	pause the target on this pod for 30 s and keep attempts. The rest of that target's batch is skipped; other targets keep running	fast_lane.worker.ts:98-102,206,218, target_pause.ts:32-55	fast_lane.worker.spec.ts:205,235; kafka.target.spec.ts:130,471
a target is paused	let one probe read through every 30 s, only on the lane the target runs on	target_pause.ts:47-55	kafka.target.spec.ts:205,273
a fast-lane row of key K is waiting for a retry	hold every later version of K on the same target. A DEAD row holds nothing	fast_lane.worker.ts:173-183,220	fast_lane.worker.spec.ts:148,162
a row succeeded	delete it	fast_lane.worker.ts:91-96	fast_lane.worker.spec.ts:182
an at-most-once send got no answer (Kafka unknown)	DEAD at once, never resent, target paused	fast_lane.worker.ts:74-77, kafka.target.ts:87	kafka.target.spec.ts:235,351
Kafka answered REQUEST_TIMED_OUT / NETWORK_EXCEPTION	at-most-once → DEAD, target not paused; at-least-once → kept, attempts 0, target paused	kafka.target.ts:46,89, fast_lane.worker.ts:76-77	kafka.target.spec.ts:415,434
the Kafka client rejects the record (for example an empty value)	DEAD at once as "not-sent, invalid record", whatever the delivery mode	kafka.target.ts:45,91, fast_lane.worker.ts:72-73	kafka.target.spec.ts:444,451
a fast-lane owner dies with a row in flight	at-most-once → DEAD "in flight when its partition was taken over"; at-least-once → resent. A row whose outcome was already written under that token stays open	fast_lane.worker.ts:249-264	kafka.target.spec.ts:177,186,300,399; fast_lane.worker.spec.ts:310
a pod lost a partition lease between rows	send no further row of that partition	fast_lane.worker.ts:205,208-211, partition_lease.service.ts:109-117	fast_lane.worker.spec.ts:260; kafka.target.spec.ts:364
a row was closed by a takeover while this pod still holds the lease	not send it	fast_lane.worker.ts:213-215	kafka.target.spec.ts:388
an exclusive code (every schedule) has a RUNNING row with a live lease	keep its other slots PENDING. Only exclusive targets wait on a busy code	long_lane.worker.ts:70,77-83, unique index jobs.schema.ts:93-96	schedules.service.spec.ts:234,262,277,356
a long-lane row is RUNNING and its lease expired	let any pod reclaim it and run it again from the start	long_lane.worker.ts:71	long_lane.worker.spec.ts:98
a schedule slot's first run starts more than 60 s after its slot time	policy skip → skip with reason previous_slot_running; run_once → skip only when a newer PENDING slot exists, otherwise run. Either skip records last_skipped_at	schedules.service.ts:169-185,188-191	schedules.service.spec.ts:234,352,354
a tick finds several missed slots, or one slot later than grace	log schedule missed. Under skip + late, enqueue nothing; else enqueue only the newest slot. Then move next_run_at past now	schedules.service.ts:139-167	schedules.service.spec.ts:163
an admin disables a schedule	create no new slots; a slot already enqueued still runs	schedules.service.ts:107	schedules.service.spec.ts:182
admin retry or skip hits a row that is not DEAD	409	jobs.controller.ts:57,66,74	jobs.controller.spec.ts:71
a cron is malformed (not 5 fields, out of range, never fires within 5 years)	throw invalid_cron (400)	cron.ts:14,29-31,47,86	schedules.service.spec.ts:68,213
JOBS_WORKERS=off	run no fast lane, long lane or tick on this pod. Schedule sync still runs	jobs.constants.ts:40, fast_lane.worker.ts:144, long_lane.worker.ts:50, schedules.service.ts:69	untested

6. States and flows

Job row states

from	event	to	code
—	enqueue (must-not-lose)	PENDING, attempts 0, next_run_at now	side_effect.service.ts:64-78
PENDING (fast)	handler ok	deleted	fast_lane.worker.ts:93
PENDING (fast)	answered failure, attempts < max	PENDING, attempts+1, next_run_at + backoff	fast_lane.worker.ts:113
PENDING / RUNNING	answered failure at max, dead_now, or at-most-once with unknown outcome	DEAD, closed_at	fast_lane.worker.ts:110-112
PENDING (fast)	unreachable	PENDING, unchanged	fast_lane.worker.ts:98-102
PENDING (fast)	owner died with the row in flight, at-most-once	DEAD	fast_lane.worker.ts:259-262
PENDING (long)	reserve	RUNNING, claim_token, lease now + 5 min, running_code when exclusive	long_lane.worker.ts:87-99
RUNNING	ok	deleted	long_lane.worker.ts:114 + fast_lane.worker.ts:93
RUNNING	answered failure or unreachable	PENDING with the claim unset (RELEASE)	long_lane.worker.ts:24-27
RUNNING, lease expired	reserve by any pod	RUNNING under a new token	long_lane.worker.ts:71
DEAD	admin retry	PENDING, attempts 0, next_run_at now	jobs.controller.ts:49-56
DEAD	admin skip	deleted	jobs.controller.ts:65

Fast-lane rows never become RUNNING. "Running" for them is the partition's inflight mark.

Enqueue

Fast lane poll

Partition lease heartbeat (every 10 s)

Fair share is ceil(64 / live pods), or max(floor, 1) while another live pod is below the floor (partition_lease.service.ts:80-96).

Long lane

Schedule tick and slot run

Schedule row

state	how it changes	code
inserted (enabled, next_run_at = next slot)	first boot that declares the code	schedules.service.ts:84-97
due	next_run_at <= now	:107
advanced	tick sets next_run_at = cron.next(now)	:162-165
disabled	admin; the tick ignores it	:120-125
re-enabled	next_run_at = cron.next(now), slots missed while disabled are dropped	:123
overridden	cron_override set; _cronOf prefers it	:127-131,193-195

7. Jobs, schedules, queues, events

Every registered job target and schedule in the code

name	lane	handler file	trigger (who enqueues)	retry	delivery / idempotency
kafka:lastmiles.oms.do-update (code do_update_oms)	fast	src/kafka/kafka.target.ts:70-94	DO create (delivery_order.service.ts:120) and DO status change (:152-153) when chanel_code ∈ DONJAI, CHOKCHAI, HLH (:27,173)	6 runs (first + 1/2/4/8/16 min), then DEAD	at_most_once; dedupe do_update_oms:<delivery_no>:<version> (:190); key = delivery_no; header event_id = the payload's event id (:183,191)
kafka:lastmiles.staffapp.do-update (code do_update_staffapp)	fast	same	same, chanel_code ∈ SPO, OD, HLH (:28,174)	same	same pattern
kafka:lastmiles.staffapp.load-update	fast	same	none: registered from KAFKA_EVENTS (kafka.constants.ts:20), no caller enqueues it	same	at_most_once
kafka:lastmiles.staffapp.task-update	fast	same	none (kafka.constants.ts:21)	same	at_most_once
kafka:lastmiles.staffapp.driver-update	fast	same	none (kafka.constants.ts:22)	same	at_most_once
kafka:<topic> from KAFKA_TARGET_TOPICS	fast	same	none in code (env-declared, kafka.target.ts:11-16,64-67)	same	at_most_once, or at_least_once with the :at_least_once suffix
schedule:<code>	long, exclusive	src/schedules/schedules.service.ts:59-64,169-185	tick, from a declare() call	retry_count total runs (default 3) at 1, 2 min	dedupe <code>:<slotISO>
lossy targets	Bull	src/jobs/lossy.processor.ts	—	1 try, dropped	none

No schedule is declared in this build. demo.schedules.ts was removed in 3dd1c5a0, and no call to SchedulesService.declare exists in src. No lossy target is registered either.

Kafka send details: the topic is the setting's topic_name override when the job's code is a declared producer event, else the target's topic (kafka.target.ts:81, kafka.config.ts:35-39). The value is JSON.stringify(payload). Headers are event_id = job._id, overridden by job.headers, plus traceparent and trace_id from the job context (kafka.target.ts:83, kafka.producer.ts:29-32,62). Error classes come from kafkaFailure (kafka.target.ts:40-51):

class	example	outcome
answered	non-retriable protocol error, a non-Kafka error	retry path
not_sent	connect phase failed, ECONNREFUSED, broker or metadata not found, NOT_WRITTEN_TYPES	TargetUnreachable → pause
unknown	send phase failed after a possible append	TargetOutcomeUnknown
timed_out	REQUEST_TIMED_OUT, NETWORK_EXCEPTION	TargetOutcomeUnknown + target_up
invalid	KafkaJSNonRetriableError raised by the client	dead_now

Every run writes one log line, job run code= target= dedupe_key= attempts= duration_ms= result= (fast_lane.worker.ts:85-87). DEAD writes job dead alert ... at error level (:119,263). The alert is that log line only.

8. Settings and env

key	default	effect	read at
JOBS_WORKERS (env)	on	off = this pod only enqueues: no fast lane, long lane or schedule tick	jobs.constants.ts:40
KAFKA_TARGET_TOPICS (env)	empty	Extra kafka:<topic> targets, topic[:at_least_once] separated by commas; can switch a declared topic to at-least-once	kafka.target.ts:64
QUEUE_REDIS_URL (env)	—	Bull connection for the lossy queue	app.module.ts:40-42
Kafka setting record, per-event topic_name	default_topic	Topic override for DO-update jobs	kafka.config.ts:35-39
JOBS_TEST_MONGO_URI (env, tests only)	mongodb://localhost:27017/?replicaSet=rs0&directConnection=true	Mongo for the jobs specs (a replica set is needed for transactions)	fast_lane.worker.spec.ts:14 and the other specs
constants (code only)	partitions 64, fast batch 100, long batch 10, retries [1,2,4,8,16] min, partial poll 1 s, idle 1/2/5 s, partition lease 30 s / heartbeat 10 s, long lease 5 min / heartbeat 1 min, probe 30 s	—	jobs.constants.ts:17-34
schedule constants	tick 15 s, grace 60 s, offset +420 min, default retry 3, skip-by-default when interval ≤ 5 min	—	schedules.constants.ts:6-14
Kafka client	connect 3 s, request 10 s, retries 0	Caps one send well under the 30 s partition lease	kafka.producer.ts:15-16

No jobs setting lives in the settings module. Retry, lease and batch numbers cannot be changed without a deploy.

9. Concurrency and recovery

situation	what the code does	code
several pods boot on an empty DB	Each pod removes duplicate partition rows, builds the unique index (retried up to 5 times), then upserts 64 rows; duplicate-key errors are ignored	partition_lease.service.ts:37-62,128-144
pod joins or leaves	The heartbeat rebalances to the fair share. Busy partitions are never released. One atomic findOneAndUpdate per take, so two pods never hold one partition	partition_lease.service.ts:65-97,157-175
pod dies mid fast-lane batch	Its leases expire after 30 s and a live pod takes them. Only the row in flight is judged: at-most-once → DEAD, at-least-once → stays PENDING and is resent. Rows not yet started stay PENDING	fast_lane.worker.ts:198-222,249-264
pod stalls past its lease, then wakes	setInflight fails on the lease guard, so the pod drops the partition and sends no more of it. A late outcome write still needs status: PENDING and the old attempts	partition_lease.service.ts:109-117, fast_lane.worker.ts:114-118
pod dies mid long-lane row	The lease expires within 5 min of the last heartbeat; any pod reclaims the row and runs it from the start. The handler may therefore run twice: long lane has no at-most-once mode	long_lane.worker.ts:71,132-139
two pods reserve long rows at once	updateMany repeats the free filter, so each row ends with one token. Exclusive codes are guarded by the unique running_code index	long_lane.worker.ts:86-104, jobs.schema.ts:93-96
same slot ticked by every pod	One row per dedupe_key; next_run_at advance is guarded on its old value	schedules.service.ts:154-165
the same business event enqueued twice	Filtered before the insert, or swallowed as a duplicate key outside a transaction	side_effect.service.ts:94-111
an inbound Kafka message delivered twice	DO side: event_key on activity logs rejects the replay, so no second job; the dedupe key repeats anyway	delivery_order.service.ts:150,190
Kafka broker down for days	Unreachable → paused, attempts kept, rows pile up PENDING; one probe every 30 s per pod; drains in key order when back	target_pause.ts:47-55, kafka.target.spec.ts:205
a downstream rejects the same key forever	That key is held until its row goes DEAD (about 31 min), then later versions flow	fast_lane.worker.ts:173-183
graceful shutdown (SIGTERM)	The fast lane stops polling, waits for the current tick, releases all partitions and deletes its job_workers row. The long lane waits for its batches. Kafka disconnects in onApplicationShutdown, after every onModuleDestroy	fast_lane.worker.ts:151-162, long_lane.worker.ts:56-62, kafka.service.ts:77-80, main.ts:52
a pod's pause map is lost on restart	Harmless: rows and attempts live in Mongo	target_pause.ts:26
old rows of another shape (no lane / partition)	Never read: both lanes filter on lane	fast_lane.worker.ts:170, long_lane.worker.ts:71, delivery_order.service.spec.ts:158

10. Web (store-v2)

store-v2 has no jobs page and no schedules page for this engine. D-214 parked the admin UI.

What exists is a page from the earlier kafka-outbound prototype (store commit e8c6338):

item	path (store-v2)	note
page /setting/outbound-events	src/pages/setting/outbound-events/index.tsx:20-174	Filters key, topic, status; a Retry button on DEAD rows; paging
menu entry "Outbound Events"	src/constants/menu.ts:102-106	No roles on the entry
Next proxy routes	src/pages/api/outbound_events/index.ts, [id]/retry.ts	Forward to API_GATEWAY + /outbound_events (src/apis/server/outbound_event/index.ts:17-23)
client types	src/apis/client/outbound_event/types.ts:1-30	Expect {items, total, page, limit} and rows with id, topic, event

This page is broken against core at 3d70475a. Core serves no /outbound_events route; the controller is jobs (jobs.controller.ts:23-26). The shapes do not match either: core returns data, not items; lean rows carry _id, not id; and core has target, not topic. The page has no Skip action and no schedules view.

11. Tests

All jobs specs need a Mongo replica set (JOBS_TEST_MONGO_URI). Run them with --maxWorkers=3.

spec	covers
src/jobs/side_effect.service.spec.ts	Partition hash; one insertMany in the session; dedupe inside and outside a transaction; abort → 0 rows; lossy only after commit; key absent; unknown target; inTransaction hook order
src/jobs/fast_lane.worker.spec.ts	Poll delays; 250 rows → 100/100/50; backoff and DEAD at the 6th failure; key hold and order; success deletes; unreachable pause over 2 days; lost lease; 3 pods → 21–22 partitions each; no unsent row closed on takeover
src/jobs/partition_lease.service.spec.ts	64 rows with 2 pods; 32/32 split; dead-pod takeover with inflight; busy never released; seeding race without an index; duplicate repair; restart is a no-op
src/jobs/long_lane.worker.spec.ts	Reserve ≤ 10 with one token and 5-min lease; 7-min run never reclaimed; reclaim after pod death; concurrent reserve; backoff and DEAD at the per-code count; unreachable; taken-over pod cannot finish
src/jobs/jobs.controller.spec.ts	Admin only; list filters and paging without payload; retry DEAD; 409 non-DEAD; 404; skip; late retry of an older version
src/schedules/schedules.service.spec.ts	Cron at +07:00; bad cron; default policy; one job per slot over 2 pods; retry inside a slot; DEAD slot then next slot; missed slots after downtime; disable; override; 404/400; one RUNNING per code (S-18); run_once collapse (S-19); other codes keep running while one exclusive code runs long (S-21/S-22)
src/kafka/kafka.target.spec.ts	Key, headers, topics; delivery-mode config; every Kafka error class; outage drain; takeover at-most-once and at-least-once; hung broker; stall past lease; invalid records; target owned by the Kafka module
src/delivery_order/delivery_order.service.spec.ts:104-235	DO change → one job → one Kafka message; duplicate DO update; broker down during the emit; old-shape rows ignored; wire contract (event_id, trace_id, bytes); topic override; rollback leaves no job

Untested: the JOBS_WORKERS=off switch, the lossy Bull processor end to end, the HTTP 403 for non-admin (only the role metadata is asserted), and any call from store-v2.

12. Gaps and known issues

#	gap	code vs design	risk
G1	store-v2 has no working jobs UI. The outbound-events page calls the missing /outbound_events with the wrong shape, and there is no schedules page	design parked the UI (D-214); the stale page remains in the menu	Admins see an error page; DEAD rows can only be handled through the API
G2	No cancel or skip of PENDING or RUNNING rows, no job detail endpoint, and the payload is never returned	the brief asks for list/retry/cancel; built: list/retry/skip-DEAD	A poisoned PENDING row can only wait for DEAD
G3	DEAD rows never expire; job_stats (spec SC2, AC-08) was built and then removed in 3dd1c5a0; the DEAD TTL (30 d, disc:D-270) is not built; the alert is a log line only	design: TTL via expireAt('jobs_dead') (log-setting D-318)	jobs grows with DEAD rows; there are no hourly counts and no paging alert
G4	No schedule is declared and 3 Kafka targets (load/task/driver-update) have no producer	the engine is ready, its users are not	The schedules path runs only in tests
G5	The Kafka send path does not check the setting enabled flag. send connects lazily with the stored brokers (kafka.producer.ts:55-80,106-121)	—	With Kafka disabled in settings, jobs still send when brokers are set, or pile up paused when none are
G6	The long lane ignores delivery_mode. After lease expiry or a long stall, the handler runs again even if the first run finished its side effect	design states at-most-once/at-least-once for the fast lane only	Long-lane and schedule handlers must be idempotent
G7	retry_count means total runs (3 = 1 run + 2 retries at 1, 2 min). It is keyed by code, so any job with the same code shares it (job_targets.registry.ts:49-51). Admins cannot change it; boot sync overwrites it	design said "retry 3 inside the slot"	Naming invites off-by-one changes
G8	A slot skipped at tick time (late, skip) is only logged; last_skipped_at is not updated. The run-time skip reason previous_slot_running is also given when the real cause was pods down or a busy lane	—	Schedule history is incomplete
G9	JOBS_WORKERS=off does not stop the Bull LossyProcessor. The lossy path has no retry and no registered user	—	Only matters once a lossy target appears
G10	The fast-lane read sorts on {key, version, _id}, which is not index-backed, and the hold query has no limit. payload.slot (run_once collapse) and the admin list filters have no index; countDocuments scans	—	Polls get slower as a multi-day backlog grows
G11	The cron parser is fixed at +07:00 and supports only *, lists, ranges and steps (no names, L, ? or seconds)	D-205 (own parser, no dependency)	—
G12	Spec paths drifted: the Kafka target is src/kafka/kafka.target.ts, not src/jobs/targets/; the generic nextVersion(key, tx) counter (disc:D-265) is not built; DO keeps its own job_counters	—	—
G13	Pause state is per pod: each pod probes a down target on its own (D-195)	as designed	N pods make N probes per 30 s

13. How to add the next one

A. A new job target (side effect that must not be lost)

1. In your module, import JobsModule (src/kafka/kafka.module.ts:18 is the pattern). The engine must import no business module; your module owns its target.
2. Make a provider that implements OnModuleInit and injects JobTargetsRegistry. Call registry.register({ name: '<area>:<thing>', lane: JobLane.fast | JobLane.long, handler, delivery_mode? }). Copy src/kafka/kafka.target.ts:53-77.
3. Write the handler so that it:
   • throws TargetUnreachable when nothing reached the target (connect refused, 5xx, client timeout before send);
   • throws TargetOutcomeUnknown when the request may have landed with no answer;
   • throws an error with dead_now: true when a resend can never succeed;
   • throws anything else for an answered failure;
   • caps each call well under 30 s on the fast lane (kafka.producer.ts:15);
   • is idempotent on the long lane (G6).
4. Enqueue inside your business transaction. Copy delivery_order.service.ts:112-122,168-195: inTransaction(this.model.db, async (tx) => { ...write...; await sideEffect.enqueueMany([{ target, code, key, version, dedupe_key, payload, headers }], tx) }).
   • dedupe_key must be derived from the business change (for example <event>:<key>:<version>), never left random, or a replayed request makes a second job.
   • Use key + a growing version when receivers need per-entity order.
5. Add a spec next to your target. Copy the fixtures in src/kafka/kafka.target.spec.ts (real Mongo replica set, injected clock, fake handler): ok, answered → DEAD after 6, unreachable → paused with attempts 0, takeover.

B. A new Kafka topic

• Producer event: add {event, type: produce, default_topic} to KAFKA_EVENTS (kafka.constants.ts:16-23). The target kafka:<default_topic> registers itself, and the per-event topic_name setting can override the topic. Enqueue with target: kafkaTargetName(default_topic) and code: <event>.
• Ad-hoc topic: set KAFKA_TARGET_TOPICS=topic[:at_least_once].

C. A new schedule

1. Import SchedulesModule in your module.
2. In a provider's onModuleInit, call schedules.declare({ code: 'snake_code', cron: 'm h dom mon dow', handler: async ({ slot, job }) => {...}, missed_slot_policy?, retry_count? }). The cron is read in +07:00.
3. The handler runs on the long lane, one RUNNING slot per code across pods. It must be idempotent per slot, because a reclaim reruns it.
4. Boot upserts the schedules row. Admins then enable or disable it and set cron_override through /api/v1/schedules.
5. Add tests modelled on src/schedules/schedules.service.spec.ts:88-220: 2 pods → 1 job per slot, retry within a slot, missed-slot policy.

D. A new lane-level behaviour

Change JobOutcome (fast_lane.worker.ts:49-122), never a second outcome path. Both lanes share it through the guard and release arguments.