gizlang:worker

giz's inference worker protocol, version 1 (giz's ml/worker and wgp-api speak it): a service takes jobs over HTTP on a thread of its own, the caller runs them one at a time where its models are, and the worker keeps each result until giz acknowledges it.

POST /v1/jobs takes {operation, model, input, idempotencyKey} as JSON, or as the form field "request" with the files it names ({"$file": field}), which stream to disk as they arrive; the executor sees each such file as {"$inputFile": {name, contentType, path}}. A key gives the same job for the same request (the first 32 hex digits of SHA-256 of model, operation and key). The worker holds capacity jobs (queued and processing) and answers 429 WORKER_BUSY beyond them. GET, DELETE /v1/jobs/{id} read and cancel a job, POST /v1/jobs/{id}/ack frees a finished one, GET /files/{id}/{name} serves its outputs. GET /v1/queue, /v1/health, /v1/models and /metrics describe the worker; POST /v1/intake/close and /v1/intake/open stop and start new jobs. Each job lives in its own folder under the state folder (job.json, its inputs and outputs), so a restart keeps the queued and finished ones; a job that was processing fails with WORKER_INTERRUPTED. Under serve, where GIZ_WORKER_AUTHORIZATION is set, every /v1/ request must carry it as its Authorization header (401 otherwise).

import * as Worker from "gizlang:worker";

Types

Model

type Model = { id: string; operation: string; }

A model the worker serves, with its operation ("video.generate").

• id: string
• operation: string

Config

type Config = { models: Vec<Model>; capacity: i64; state: string; }

The worker: its models, the jobs it holds at once, and the folder of its jobs.

• models: Vec<gizlang_worker.Model>
• capacity: i64
• state: string

Work

type Work = { id: string; operation: string; model: string; input: JsonValue; dir: string; }

A job for the executor: its input (files as {"$inputFile": ...}) and the folder its outputs go into.

• id: string
• operation: string
• model: string
• input: gizlang_json.Json
• dir: string

Failure

type Failure = { code: string; message: string; status: i64; }

Why a job failed: its code (WORKER_INVALID_INPUT, or the executor's own), message and HTTP status (4xx: the request; 5xx: the worker).

• code: string
• message: string
• status: i64

Progress

type Progress = { stage: string; step: i64; total: i64; remaining: f64; }

Where a job is: its stage as giz names them (initializing, loading_model, encoding_prompt, generating, encoding_video), its step of total, and the seconds it expects to take still.

• stage: string
• step: i64
• total: i64
• remaining: f64

Record

type Record = { id: string; operation: string; model: string; input: string; status: string; stage: string; step: i64; total: i64; remaining: f64; output: Vec<string>; error: string; code: string; errorStatus: i64; created: i64; started: i64; completed: i64; acknowledged: bool; cancel: bool; }

One job's state. Times are milliseconds since 1970 (0: not yet).

• id: string
• operation: string
• model: string
• input: string
• status: string
• stage: string
• step: i64
• total: i64
• remaining: f64
• output: Vec<string>
• error: string
• code: string
• errorStatus: i64
• created: i64
• started: i64
• completed: i64
• acknowledged: bool
• cancel: bool

State

type State = { sys: System; config: Config; jobs: Task.Cell<Vec<Record>>; lock: Task.Atomic; accepting: Task.Atomic; wake: Task.Channel<i64>; done: Task.Atomic; failed: Task.Atomic; aborted: Task.Atomic; busyMs: Task.Atomic; port: Task.Atomic; closing: Task.Atomic; authorization: string; }

What the server's thread and the executor share: the jobs (changed only under lock), whether intake is open, the wake-up of the executor, the counts of finished jobs and their milliseconds, the port the server took (0 until it listens, -1 when it cannot) and whether it is to stop.

• sys: gizlang_system.System
• config: gizlang_worker.Config
• jobs: gizlang_task.Cell<Vec<gizlang_worker.Record>>
• lock: gizlang_task.Atomic
• accepting: gizlang_task.Atomic
• wake: gizlang_task.Channel<i64>
• done: gizlang_task.Atomic
• failed: gizlang_task.Atomic
• aborted: gizlang_task.Atomic
• busyMs: gizlang_task.Atomic
• port: gizlang_task.Atomic
• closing: gizlang_task.Atomic
• authorization: string

Worker

type Worker = { state: State; server: Task.Thread<i64>; port: i64; }

A started worker: its state, the server's thread and its port.

• state: gizlang_worker.State
• server: gizlang_task.Thread<i64>
• port: i64

Functions

start

function start(sys: System, host: string, port: i64, config: Config, authorization: string = ""): Worker throws string

Serves config's models on host:port (0: a free port, Worker.port) on an OS thread of its own, so it answers while a job runs. The caller runs the jobs that next gives, on its own thread with the models it keeps loaded, tells their progress to report and their end to finish, and ends with stop. Each /v1/ request must carry authorization as its Authorization header unless it is "".

Effects: fail, read, write, console, io.

next

function next(w: Worker): Option<Work>

The next job to run, now processing; it waits for one. None once the worker stops.

Effects: fail, read, write, console, io.

report

function report(w: Worker, work: Work, p: Progress): bool

Keeps the progress of work; false once giz cancelled it, when the caller stops it and finishes it with any failure.

Effects: fail, io.

finish

function finish(w: Worker, work: Work, result: Result<Vec<string>, Failure>): void

Keeps how work ended: the names of the output files it made in work.dir, or its failure.

Effects: fail, read, write, console, io.

stop

function stop(w: Worker): void

Stops the server (requests in flight finish) and waits for its thread; next then gives none.

Effects: fail, io.

invalid

function invalid(message: string): Failure

A failure of the job's own input: WORKER_INVALID_INPUT, status 400.

failed

function failed(code: string, message: string): Failure

A failure of the worker: code (the executor's own: WORKER_FAILED, WORKER_STORAGE, ...), status 500.

progress

function progress(w: Worker, work: Work, stage: string, step: i64, total: i64): bool

Tells giz where work is: its stage, step of total. False once giz cancelled it, when the caller stops it and finishes it with failed(WORKER_CANCELLED, ...).

Effects: fail, io.

uploadedImages

function uploadedImages(sys: System, input: JsonValue, key: string): Vec<Img.Image> throws Failure

The pictures of field key of a job's input (one uploaded image, or a list of them), decoded to PICTURE_SIDE.

Effects: fail, io.

seed

function seed(sys: System, input: JsonValue, bound: i64 = 4294967296): i64

The seed of a job: its input's seed when 0 or more, else one from the clock below bound.

Effects: fail, io.

serve

function serve(sys: System, name: string, host: string, port: i64, config: Config, run: (w: Worker, work: Work) => Result<Vec<string>, Failure>, after: () => void): i64 throws string

Serves config's models on host:port, as start, and runs each job that giz sends, one at a time, with run (the executor: it tells its progress with progress and gives the names of its outputs in work.dir, or fails), then calls after (the models' tunings to save, say). Each job's end and time is printed (the service's journal). It returns when the worker stops.

Effects: fail, read, write, console, io.

inputPath

function inputPath(value: JsonValue): Option<string>

The path of an input file ({"$inputFile": {...}}) of a job's input; none for any other value.

Effects: fail.

Constants

WORKER_BUSY

const WORKER_BUSY = "WORKER_BUSY"

The error code of a job refused because the worker holds capacity jobs.

WORKER_INTERRUPTED

const WORKER_INTERRUPTED = "WORKER_INTERRUPTED"

The error code of a job that was processing when the worker stopped.

WORKER_INVALID_INPUT

const WORKER_INVALID_INPUT = "WORKER_INVALID_INPUT"

The error code of a job whose input the executor refuses.

PICTURE_SIDE

const PICTURE_SIDE = 2048

The longer side that a picture of a job is decoded to at least: a JPEG or PNG at the smallest scale (1/2, 1/4, 1/8) that keeps it, so a phone photo or a screenshot of any size is within the decoder's bounds. A model reads a picture at 4 megapixels at most, which 2048 covers.