Jobs: asynchronous workflows

Two Gradle modules, one idea

The jobs engine is two modules under jobs/ (jobs/README.md):

• jobs/proto (protomolt-jobs-proto) is the wire contract, and it contains exactly two messages: WorkflowRunEvent, the lifecycle envelope published on the events topic, and WorkflowRunRequest, the payload accepted on the request topic for broker-native submission. There is deliberately no gRPC service here: the four job verbs are descriptor-native ProtoMoltService catalog actions, and broker-native clients never touch gRPC at all.
• jobs/service (protomolt-jobs-service) is the store, the worker fleet, the outbox relay, and the four verbs. It is deliberately framework-free: plain JDBC over HikariCP with Flyway migrations, plain kafka-clients, and Java virtual threads throughout.

It mounts as the composable jobs role (JobsModule), and it requires a co-mounted registry role, because stored workflow definitions resolve through the registry's contributed WorkflowRepository. On one node the jobs role co-mounts with registry and parse; the job verbs, the parse-and-index workflow and replay-documents all ride the registry's actions route.

WorkflowRunDatabase database = new WorkflowRunDatabase(new WorkflowRunStoreConfig(jdbcUrl, user, password));
JdbcWorkflowRunStore store = new JdbcWorkflowRunStore(database);
WorkflowRunWorker worker = new WorkflowRunWorker(store, actionContext, workflowRepositoryOrNull,
        new WorkflowRunner(), config);          // config: WorkflowRunsConfig
worker.start();                              // workerCount claim loops + request-topic consumer
WorkflowRunEventRelay relay = new WorkflowRunEventRelay(store,
        WorkflowRunEventRelay.newProducer(bootstrap, schemaRegistryUrlOrNull),
        config.eventsTopic(), Duration.ofMillis(500), 100);
relay.start();

Async is an execution mode, not a dialect

Jobs does not define its own workflow language. A submitted workflow is the same descriptor-first JSON envelope the synchronous run-workflow verb accepts, compiled into the same CompiledWorkflow record (name, schema descriptor set, input type, an ordered list of WorkflowSteps, optional output mapping, whole-run deadline_ms defaulting to 30 seconds). The worker replays that definition through the same WorkflowRunner; run-workflow itself is unchanged. What jobs adds is everything around the edges: submission, claiming, checkpointing, retrying, parking, and publishing.

Submission is strict about what it persists. WorkflowRunSubmitter.submit parses and verifies the definition and validates the input against the workflow's input type before anything is written, then snapshots the workflow envelope and the input into the job row itself. The design rule: a workflow edited later never shifts a live job.

workflow.put("name", "embed-text");
workflow.putObject("schema").put("descriptorSetBase64", descriptorSetBase64);
workflow.put("inputType", "jobs.test.Text");
tokenize.put("name", "tokenize");
tokenize.put("target", target);
tokenize.put("method", "jobs.test.Tokenizer/Tokenize");
tokenize.putArray("rules").add("text = input.text").add("fail = input.fail");
review.put("name", "review");
review.put("method", "jobs.test.ReviewDesk/Review");
review.putArray("rules").add("text = input.text");
review.put("completion", "external");   // park here until a human answers

The completion: "external" marker is the whole human-in-the-loop mechanism: a step declaring it does not get invoked at all, it gets parked. The synchronous engine refuses to run such a workflow, so a parked step can never strand a held connection:

transform workflow runner: the engine will not park a synchronous run

"synchronous run-workflow cannot park - submit the workflow as a job
 (submit-workflow) so complete-step can supply the response"

A run is one Postgres row

One Flyway migration creates the workflow_run table, and one row in that table is the entire runtime state of a job: job_id (a client-generated uuid that doubles as the idempotency key), workflow_name, workflow_definition (the JSONB snapshot), input (JSONB), status, attempt and max_attempts, run_after (when the row becomes claimable), outstanding_step, checkpoints (an ordered JSONB array), result, verdict, error, lease_owner / lease_until, and timestamps. If you can read the row, you can debug the job at 2am; that is a stated design goal.

flowchart TD
          q["QUEUED: claimable once run_after is in the past"] --> r["RUNNING: leased to exactly one worker"]
          r --> c["COMPLETED: verdict carries the one-line summary"]
          r --> w["WAITING: parked on an external step"]
          w -->|"complete-step accepted"| q
          r -->|"retryable failure: run_after pushed into the future"| q
          r -->|"lease expired: the sweeper requeues"| q
          r --> f["FAILED: validation verdict, or a non-retryable error"]
          q -->|"attempts exhausted"| d["DEAD: the dead-letter state, nothing re-enqueues it"]

Many workers claim from the same table without ever taking the same job, and the entire claim is a single SQL statement: the sub-select finds the oldest eligible QUEUED row and locks it (FOR UPDATE SKIP LOCKED means a row already locked by another worker is skipped, not waited on), and the update flips it to RUNNING under that lock, stamps the lease, and increments attempt.

jobs service run store: the whole claim in one statement

UPDATE workflow_run SET status = 'RUNNING', lease_owner = ?,
       lease_until = ?, attempt = attempt + 1, updated_at = now()
 WHERE job_id = (SELECT job_id FROM workflow_run
                  WHERE status = 'QUEUED' AND run_after <= now()
                  ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1)
RETURNING ...

The lease is how a crashed worker loses its job: a sweeper (requeueExpiredLeases) flips any RUNNING row whose lease_until has passed back to QUEUED, and the next claim resumes from the checkpoints, not from step zero. The default lease is 5 minutes, sized to sit above p99 segment latency, because LLM steps take minutes, not milliseconds.

Watch one run

This is a run of a workflow shaped like the module's own test fixtures: tokenize a text, embed it, then a human reviews the result before the job closes. Press run and watch the row move. Timings are compressed (the real default backoff is 5 seconds, doubling per attempt); the mechanics are the real ones. Three things to look for: after the failed attempt, tokenize is not re-executed, its checkpoint is replayed; the backoff wait is visible because run_after is a real column; and the WAITING gate waits for you, exactly like the real job waits for a complete-step call.

The colored tags in the event log are the real event types from WorkflowRunEvent.Type. Note what is absent: there is no event for claiming or retrying. Events mark commit points (accept, step checkpoint, park, terminal state); retries and lease sweeps are just the row changing state. And notice the terminal discipline baked into the contract: a COMPLETED event must carry its verdict, and a FAILED or DEAD event must carry the verbatim error, enforced by CEL rules on the message itself.

enum Type {
  TYPE_UNSPECIFIED = 0;     // never published, so an unset type is detectable
  TYPE_ACCEPTED = 1;        // accepted and queued (the submit commit point)
  TYPE_STEP_CHECKPOINT = 2; // a step's response was checkpointed to the job row
  TYPE_WAITING = 3;         // parked on an external-completion step
  TYPE_COMPLETED = 4;       // verdict carries the one-line summary
  TYPE_FAILED = 5;          // validation verdict or non-retryable error
  TYPE_DEAD = 6;            // retries exhausted; nothing re-enqueues a DEAD job
}

Checkpoints: resume at the first missing step

Checkpointing is the whole game (jobs/README.md). Each step's response is persisted to the job row in the same transaction as its STEP_CHECKPOINT event, before the next step runs. If the worker crashes, or its lease expires and the sweeper requeues the row, the next claim replays the stored definition, rebuilds the checkpoint prefix, and resumes at the first missing checkpoint. A store failure inside the checkpoint observer requeues the job rather than settling it as failed work.

jobs/README.md: one checkpoint entry per executed step, in order; response is proto3 JSON of the step's output type

{ "name": "tokenize", "skipped": false, "response": { "ids": [4811, 9902] } }

Replay is not trusting. Every stored checkpoint re-parses against its step's declared output type, and checkpoint entry i must name step i: a workflow edited under a live job fails loud instead of resuming against a shifted scope. One corollary follows, stated verbatim in the README: side-effecting steps must be idempotent on job_id + step_name, because a retried step may have done its side effect before its checkpoint landed.

Failure taxonomy: retry, answer, or die loudly

When a step throws, the worker classifies the failure (WorkflowRunWorker.handleFailure, riding WorkflowExecutionException.kind()) into exactly three buckets:

• Retryable. A gRPC UNAVAILABLE, DEADLINE_EXCEEDED or RESOURCE_EXHAUSTED, or a whole-segment deadline. The job requeues with exponential backoff, base × 2^(attempt − 1) seconds (default base 5 seconds), until attempt == max_attempts (default 3). Then it lands DEAD with the last error verbatim. Nothing re-enqueues a DEAD job; that is operator territory.
• Verdict. A validation failure is not an error, it is the workflow's answer: the job FAILS with the violations written into the record, and the review queue consumes these events directly.
• Non-retryable. A gate, mapping or workflow defect is deterministic corruption; retrying cannot help, so the job fails loud.

Two guardrails keep the fleet from hurting itself. The per-target concurrency cap makes a worker acquire a semaphore permit (bounded at maxConcurrentPerTarget, default 8) for the target of the job's next unexecuted step before running a segment. This is a documented approximation, a segment may call several targets, but it exists because the inference tier's bottleneck is one box per model, and an uncapped worker fleet would overwhelm it. And every worker carries a required, non-blank workerId: a fleet of anonymous workers cannot be debugged from the row.

Waiting on a human

A step declaring completion: "external" parks the job: checkpoints persisted, a WAITING event written, worker stopped. Supplying the answer is the complete-step verb, and it is careful. The job's state is gated first, only a WAITING job parked on exactly that step accepts the response. The response is parsed against the step's output type and, when the step declares validate, checked against its declared rules; a rejection fails the job as a verdict. Accepted, the checkpoint appends and the job requeues in one transaction.

The gate runs twice by design: fail-fast in the action, then again under the row lock in the store, so a race between two callers answers the same way. Redelivery is idempotent through a sealed ParkedCompletion verdict, Completed / AlreadyDone / WrongState, decided atomically under the lock. Call complete-step twice and the second call gets AlreadyDone, not a duplicated step.

Events that cannot drift

Every commit point (accept, step checkpoint, park, terminal state) inserts one row into workflow_run_events_outbox in the same transaction as the job mutation, so an event can never drift from the state change it describes. A relay drains PENDING rows oldest-first with FOR UPDATE SKIP LOCKED and publishes to the workflow-run-events topic, keyed by job_id so one job's events stay partition-ordered. Delivery is at-least-once: the relay publishes first and marks the row PUBLISHED after the broker acks, so a crash mid-flight republishes on restart, and consumers dedupe on event_id (the outbox row's own id). A row that exhausts 10 relay attempts lands FAILED: that row is the dead-letter queue, and the relay deliberately never re-enqueues it. Published rows are retained, not deleted.

There is a second door in, for clients that never speak gRPC. Producing a WorkflowRunRequest (just job_id, workflow_name, and input) to the workflow-run-requests topic is exactly equivalent to calling submit-workflow with a stored name. A consumer group named protomolt-jobs-worker reads them with auto-commit off and commits each offset only after the row commits, so a crash replays the insert, which is idempotent on job_id. An unknown workflow name does not vanish into a log: the worker writes a FAILED row loudly.

The four verbs on the shared ProtoMoltService contract:

• submit-workflow: submit a workflow as a durable asynchronous job. The request enforces, via a CEL rule, exactly one definition source: an inline workflow, or a stored workflow name, never both, never neither.
• get-job: read one run by id, with checkpoints and verdict.
• list-jobs: list runs newest first, filtered by status or name. A limit of zero selects the default 50; anything above the ceiling of 500 is clamped rather than refused, because the ceiling bounds the answer, not the request.
• complete-step: complete a run's parked external step with a validated response.

A server without jobs configured still mounts all four verbs; they answer unavailable on a null store, instead of disappearing.

surface service contract: the submit rule, enforced at the message level

// exactly one definition source: inline workflow, or stored name
expression: "has(this.workflow) != (this.workflow_name != '')"

Two limits are deliberate today: payloads ride inline in the job row rather than by claim-check reference, and there is no streaming watch RPC. The events topic is the watch lane, and get-job and list-jobs cover interactive debugging. Submission is driven by callers, through the verb or the request topic; no other subsystem enqueues jobs yet.