04. Results DB & batch workflows

Phase 2 — the generic results database, scheduled and manual triggers, and batch fan-out with parent/child rows and a stateless continuation rule.

Outcome

Scheduled and on-demand workflows — nightly retrains, hourly scoring, ad-hoc backfills, large batch-inference fan-outs — all run as ephemeral ACA Jobs and record their state in one generic results DB. Batch inference tracks per-item 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. This is Phase 2 (docs/04).

## Design — the generic results DB

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 Blob
    error        TEXT NULL,
    attempts     INT NOT NULL DEFAULT 0,
    triggered_by TEXT NULL,                        -- 'schedule' or caller email (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.

Design — triggers and the continuation rule

  • Schedule — a cron expression on the Job definition; ACA starts a fresh execution, no always-on scheduler.
  • Manual — started via the ACA execution API from the dashboard or CI, always with a triggered_by caller and gated by a scoped role. Does not depend on the dashboard being up (az containerapp job start works too).
  • Event — from an event source when a use case needs it.

A linear pipeline (extract → validate → score → publish) is one script in one Job. “Run until done” is a stateless rule over the results DB: re-dispatch children still in PENDING/RETRY up to attempts cap; the batch is done when no child is non-terminal.

Build in projects/ml-platform/

projects/ml-platform/
├── src/ml_platform/results/
│   ├── schema.sql              # CREATE TABLE results + index (idempotent, IF NOT EXISTS)
│   ├── store.py                # create_run, create_children, mark, pending_children, finalize_parent
│   └── continuation.py        # run_until_done: stateless PENDING/RETRY loop + circuit breaker
├── src/batch_job/
│   ├── Dockerfile              # pinned base + deps + ml_platform (mirrors train_job)
│   ├── requirements.txt        # mlflow/sklearn/pandas pinned + psycopg + azure-identity
│   └── score.py                # load model → create parent/children → run_until_done
└── infra/
    ├── main.tf                 # + module "batch_job" (two-pass gating on batch_image)
    ├── variables.tf            # + batch_image, batch_schedule_cron
    └── modules/batch_job/      # ACA Job: id-jobs-batch, cron schedule + manual trigger
        ├── main.tf
        ├── variables.tf
        └── outputs.tf

The results/ module is new ML-platform infrastructure added in this chapter; it formalizes the results table contract that common/results.py began writing in Ch 03. The existing record_run context manager uses exactly the columns defined in schema.sql, so Ch 03 code needs no change when Phase 2 goes live.

How the pieces connect

Results module (ml_platform/results/)

schema.sql is idempotent (CREATE TABLE IF NOT EXISTS) and is run by deploy.ps1 after grants.sql — same pattern as the Postgres principals setup. It adds the parent_id foreign key and the error/attempts columns that common/results.py pre-announced but didn’t create.

store.py exposes five functions that cover the full 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 (all children terminal, none FAILURE) or FAILURE

All five functions are no-ops when PGHOST is unset, so the batch Job is runnable locally against a mocked model without 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:

  1. Fetch pending_children.
  2. For each child: call processor(child), mark SUCCESS; on BatchItemFailure mark FAILURE (permanent); on any other exception mark RETRY (transient).
  3. If no child changed state (no progress), circuit-break → mark parent FAILURE.
  4. Stop when no eligible children remain; call finalize_parent.

A crashed or re-deployed batch Job simply re-evaluates pending_children and continues from where the DB says it stopped — there is no in-memory queue to rebuild.

Batch Job (src/batch_job/score.py)

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

  1. Load a pinned model (models:/<name>/<version> or @champion).
  2. Read the input CSV; split into chunks.
  3. Create one parent row + one child row per chunk (via store.create_children).
  4. Call run_until_done; the processor scores one chunk and calls store.mark with the per-chunk output summary.
  5. Set parent output (n_chunks, total_rows, model_ref) and exit non-zero on anything other than SUCCESS.

Infra (infra/modules/batch_job/)

The module mirrors train_job exactly: azurerm_container_app_job with a manual_trigger_config block plus a dynamic "schedule_trigger_config" block gated on var.schedule_cron != "". The identity is id-jobs-batch; the env vars are the same set (AZURE_CLIENT_ID, MLFLOW_TRACKING_URI, PGHOST, PGUSER, RESULTS_DB, IMAGE_DIGEST) plus DATA_SOURCE and MODEL_NAME for the scheduled run’s default inputs. Two-pass gating: the module’s count is 0 until both batch_image and mlflow_image are set.

The Terraform wiring (infra/main.tf) passes id-jobs-batch’s identity (provisioned by foundation/identities.tf in Ch 02) and the MLflow App URL from module.mlflow_app.

## Golden-path position & acceptance evidence

This chapter builds the promote → batch Job (scheduled/manual, parent/child rows) segment and the operational backbone the whole path reports into.

Acceptance evidence:

  • A query returns full status/output/error for a run and its children.
  • A batch of N items yields one parent row + N child rows; forcing a transient failure on one item marks it RETRY and the continuation rule re-dispatches only that item until it reaches a terminal state.
  • A scheduled trigger starts a fresh execution with triggered_by='schedule'; a manual start records the caller’s identity.

## Extensions (deferred from the MVP)

Deferred Contract MVP substitute
Broker / queue backpressure for huge fan-out docs/04 upgrade Parent/child rows + continuation (see Ch 09)
Event-triggered workflows docs/04 Schedule + manual only
Cross-workflow DAGs docs/00 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.

Back to top