04. Results DB & batch workflows

The generic results database every workflow reports into, batch fan-out as parent/child rows written by a one-shot Compose container, and a stateless continuation rule instead of an orchestration engine.

This chapter builds the platform’s operational backbone: one generic results DB that records the state of every run of every type, plus the first workflow that needs it in earnest. Batch inference fans out as one parent row per batch and one child row per chunk, tracks per-chunk success/failure, distinguishes transient from permanent failures, and can run until done without any orchestration engine, because all the state it needs lives in the results DB.

On the Compose stack the consumer is a batch scoring job: a one-shot container built from src/batch_job/Dockerfile, scoring model versions registered by chapter 03. Triggers here are on-demand only, via the dashboard buttons and runner API of chapter 02; scheduled execution arrives when Part II runs this same image as an ACA Job with cron triggers (chapter 11).

Design: results state, triggers, and bounded retries

One Postgres table records the state of every job of every type:

CREATE TABLE results (
    id           TEXT PRIMARY KEY,                 -- UUID, or deterministic hash for idempotent items
    parent_id    TEXT NULL REFERENCES results(id), -- NULL = top-level run; set = child
    name         TEXT NOT NULL,                    -- workflow/task type, e.g. 'batch:score-fraud'
    status       TEXT NOT NULL,                    -- PENDING|STARTED|SUCCESS|RETRY|FAILURE|REVOKED
    output       JSONB NULL,                       -- per-task metadata; big payloads go to object storage
    error        TEXT NULL,
    attempts     INT NOT NULL DEFAULT 0,
    triggered_by TEXT NOT NULL,                    -- 'schedule' or caller identity (audit)
    created_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at   TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX ON results (parent_id, status);

RETRY means transient/retriable; FAILURE means permanent. Batch inference uses one parent row per batch and one child row per item/chunk, giving per-item success/failure without any bespoke ledger. Locally the table lives in the results database inside the Compose Postgres container (chapter 02); in Part II it moves to Azure Database for PostgreSQL unchanged.

Three ways a workflow can start:

On-demand: the dashboard’s trigger buttons or a direct runner call starts a one-shot batch process; every row records the caller in triggered_by.
Scheduled (Part II): the same image runs as an ACA Job whose definition carries a cron expression; each tick starts a fresh execution with triggered_by='schedule' (chapter 11).
Event: started from an event source when a use case needs it. This remains deferred because the baseline has no event-driven requirement.

A linear pipeline (extract → validate → score → publish) is one script in one container. Inside an execution, run until done means: re-dispatch children still in PENDING or RETRY up to the attempt cap; finish when no child is eligible. Settled child rows survive each retry loop iteration.

The boundary is important: score.py creates a fresh parent ID when a new process starts unless the caller deliberately supplies RESULTS_RUN_ID. The baseline therefore does not promise automatic resume after a container or ACA execution crashes. Cross-execution resume would also need a stable parent ID and a pinned input snapshot; that machinery is deferred until a real workload requires it.

Build in `projects/ml-platform/`

projects/ml-platform/
├── src/ml_platform/results/
│   ├── schema.sql              # results DDL, owned once: CREATE TABLE IF NOT EXISTS + index
│   ├── store.py                # create_run, create_children, mark, pending_children, finalize_parent
│   └── continuation.py         # run_until_done: PENDING/RETRY loop + circuit breaker
├── src/batch_job/
│   ├── Dockerfile              # pinned base + deps + ml_platform package (same pattern as train_job)
│   ├── requirements.txt        # mlflow/sklearn/pandas pinned + psycopg (+ azure-identity for Part II auth)
│   └── score.py                # entrypoint: load model → create parent/children → run_until_done
└── demo/
    ├── docker-compose.yml      # + one-shot batch service; BATCH_CHUNK_SIZE pinned to 100
    └── postgres/init/01-create-results.sql  # creates the results db and applies the DDL at first boot

The results/ module is environment-neutral. It connects through ordinary PG* variables, so the same code writes to the Compose Postgres container today and to Azure Database for PostgreSQL in Part II; only connection settings differ (local password auth versus short-lived Entra tokens from a managed identity). The DDL is owned once by schema.sql: the Compose init script materializes the identical table definition when Postgres first initializes its volume, and the Part II deploy scripts apply the same file after grants.sql, so both environments converge on one schema.

This module formalizes the table contract that common/results.py began writing in chapter 03: the record_run context manager delegates to this store, so training rows and batch rows share one writer, one status vocabulary, and one table from here on.

How the pieces connect

Results module (`ml_platform/results/`)

schema.sql owns the DDL once, and every environment applies it the same way. Locally, demo/postgres/init/01-create-results.sql creates the results database and executes the identical CREATE TABLE IF NOT EXISTS statements; the Postgres image runs init scripts only when its data volume is empty, so a docker compose down -v reset rebuilds the schema. In Part II the deploy scripts apply schema.sql after grants.sql.

store.py exposes five functions that cover the lifecycle:

Function	Purpose
`create_run(name, triggered_by=…)`	Insert a top-level row in PENDING
`create_children(parent_id, items, name=…, triggered_by=…)`	Bulk-insert child rows with deterministic ids (`SHA-256(name:item_key)[:32]`); idempotent, `ON CONFLICT DO NOTHING`
`mark(run_id, status, output=…, error=…, increment_attempts=…)`	Update a row’s status and metadata
`pending_children(parent_id, max_attempts=…)`	Query PENDING/RETRY children below the attempt cap
`finalize_parent(parent_id)`	Set parent SUCCESS when every child succeeded, otherwise FAILURE

All five are no-ops when PGHOST is unset, so any job linked against the store stays runnable outside a live database.

Continuation rule (`ml_platform/results/continuation.py`)

run_until_done(parent_id, processor, max_attempts=3, max_iterations=10) applies the rule in a loop:

Fetch pending_children; each attempt increments the child’s attempts.
For each child, call processor(child). Mark normal completion SUCCESS, a BatchItemFailure as permanent FAILURE, and another exception as RETRY.
If no child changes state, circuit-break and mark the parent FAILURE.
Stop when no eligible children remain; call finalize_parent.

This gives bounded retries and idempotent child creation within one batch execution. A fresh invocation starts a fresh parent unless its caller reuses RESULTS_RUN_ID; the current dashboard and ACA schedule do not implement that cross-execution resume protocol.

Batch scoring (`src/batch_job/score.py`)

The entrypoint is symmetric to train.py but read-only to MLflow:

Load a pinned model. --model-version names an exact version; omitted, it uses the production alias selected by the promotion path.
Read the input CSV and split it into --chunk-size rows.
Create one parent row plus one child row per chunk.
Call run_until_done; the processor scores one chunk and records its output.
Record the parent summary and exit non-zero on anything other than SUCCESS.

Trigger path

At Compose startup the one-shot batch service waits for the bootstrap training run and scores the sample CSV once. On-demand calls then use the dashboard’s Run batch scoring button or POST /api/runs/batch/trigger. The runner validates scalar parameters, launches score.py, and returns immediately. It passes the execution name as RESULTS_RUN_ID, so that execution ID is also the parent results-row ID.

Part II runs the same image as the batch ACA Job. Terraform adds manual and optional cron triggers under id-jobs-batch; the dashboard starts independent executions and can supply per-execution CLI arguments without changing the Job definition.

Golden-path position & acceptance evidence

This chapter adds the registered version → batch scoring → per-chunk results segment of the golden path and builds the operational backbone that every later chapter reports into.

Acceptance evidence:

One query returns full status, output, and error data for a run and its children.
A batch of N chunks yields one parent row plus N child rows. A transient child failure is retried only up to the configured cap during that execution.
Permanent failures and exhausted retries finalize the parent as FAILURE; the circuit breaker bounds a continuation loop that stops making progress.
An on-demand trigger records its caller in triggered_by, and the local trigger response’s execution ID is the parent row ID.
Starting a new batch without the same RESULTS_RUN_ID creates a new parent. Automatic crash recovery across separate executions is explicitly not part of the baseline.
Scheduling is out of scope for Compose. ACA cron triggers start this same image in chapter 11.

Extensions (deferred from the MVP)

Deferred capability	Current baseline
Event-triggered workflows (revisit when a use case needs reactive starts or queue backpressure)	On-demand triggers now; cron via ACA Job schedule triggers in Part II (chapter 11)
Cross-workflow DAGs (revisit when pipelines need inter-job dependencies)	One script per job; no orchestration graph

Next: 05 — Online serving & promotion puts a model version behind an HTTP endpoint with version-based promotion and rollback.

Design: results state, triggers, and bounded retries

Build in projects/ml-platform/