Async Job API

Category: Engineering & DevOps

Get this pack →

This page is generated from the Air Pipe marketplace. Browse it live to install into your organization.

Expose your own long-running work as an async API. The caller POSTs, gets a job id back in milliseconds, and the answer is pushed to a realtime channel when it lands — with a status route as the fallback for whoever missed the push.

Self-hosted only for now. respond: early is honoured by the self-hosted engine. Managed does not check the key yet, so there the interface runs synchronously and the caller waits — the one thing this pack exists to avoid. Nothing errors, so the only symptom is a slow request.

The problem

A request that takes 90 seconds does not fail because the work is hard. It fails because something between you and the caller stops waiting: Cloudflare cuts a proxied request at ~100s and returns a 524, most load balancers default to 60s, and a phone changing network drops it sooner than either. The work usually finishes — the result just has nowhere to go, because the only place it ever existed was that connection.

Push first, poll as the fallback

POST /jobs-submit   ──►  row written, {id} returned          ~4ms
                         work runs detached, nobody waiting
     ws  jobs:alice ◄──  {"job_id": "...", "status": 200}    the moment it lands
POST /jobs-status   ──►  the row                             only if you missed the push

A client holding the socket learns the outcome immediately. Polling exists for the reload, the cold start, and the client that cannot hold a socket. Building it the other way round — poll first with push bolted on — gives you the latency of polling and the complexity of both.

What's included

Interface	Method	What it does
jobs-init	POST	Creates the table. Run once; idempotent
jobs-submit	POST	Auth, record the job, start it, return the id
jobs-run	POST	respond: early — does the work, records the outcome, pushes it
jobs-status	POST	Reads a job. The fallback path
live	WS	Subscribe to jobs:<owner> for completions

Setup

Variable	What it is
JWT_SECRET	Signs and verifies the caller's token
SELF_URL	This deployment's base URL, so submit can start the runner
WORK_URL	The endpoint doing the actual work — replace with yours

SQLite by default so it runs with nothing else installed — this pack exists to show the pattern with no setup. For a deployment that runs more than one job at a time use Async Job API (Postgres): SQLite is single-writer, and every detached run writes a row the moment its work finishes, so concurrent jobs contend on the writer lock rather than on anything you control.

# 1. once
curl -X POST https://<host>/jobs-init -d '{}'

# 2. start a job. `owner` comes from the token, so the body carries only the work.
curl -X POST https://<host>/jobs-submit -H "airpipe-jwt: $TOKEN" \
     -H 'content-type: application/json' \
     -d '{"payload":{"report":"quarterly"}}'
# -> { "accepted": true, "id": "...", "status": "running", "watch": "jobs:alice" }

# 3. the answer arrives on the channel. This is only the fallback:
curl -X POST https://<host>/jobs-status -H "airpipe-jwt: $TOKEN" \
     -H 'content-type: application/json' -d '{"job_id":"..."}'

The token

This pack verifies tokens; it does not issue them — your own login does. A token must be signed HS256 with JWT_SECRET and carry an owner claim, which is the identity everything here hangs off: the job is filed under it, the channel is jobs:<owner>, and the status route only returns that owner's jobs. It is never read from the request body, so a caller cannot file a job as somebody else.

If you have no login yet, the Streaming Chat pack's session interface shows the minting side (add_jwt), or for a throwaway one:

python3 - <<'PY'
import base64, hmac, hashlib, json, time
SECRET = "your-JWT_SECRET"
b = lambda x: base64.urlsafe_b64encode(x).rstrip(b'=').decode()
h = b(json.dumps({"typ":"JWT","alg":"HS256"}, separators=(',',':')).encode())
p = b(json.dumps({"iss":"Air Pipe","exp":int(time.time())+3600,"owner":"alice"},
                 separators=(',',':')).encode())
print(f"{h}.{p}." + b(hmac.new(SECRET.encode(), f"{h}.{p}".encode(), hashlib.sha256).digest()))
PY

Watching the channel

const ws = new WebSocket(`wss://<host>/ws/<ORG_UUID>/production/live?token=${token}`);
ws.onopen = () => ws.send(JSON.stringify({ ap_subscribe: `jobs:${owner}` }));
ws.onmessage = (e) => {
  const { data } = JSON.parse(e.data);          // { job_id, status }
  if (data?.job_id) finished(data.job_id, data.status);
};

<ORG_UUID> is your organisation id; self-hosted single-tenant uses 00000000-0000-0000-0000-000000000000. Subscribing to another owner's channel is refused by authorize-owner, so the token is what decides, not the channel name you ask for.

Three things worth copying

Auth cannot live on the detached interface. respond: early commits the status code before any action runs, so an interface carrying it answers 202 to an invalid token. That is why this is three interfaces: jobs-submit authenticates synchronously, and only the work detaches.

Name channels from a bounded set. jobs:<owner>, never jobs:<job_id>. The broker keeps a commit log per subscribed filter and never releases it, so a channel named from a fresh job id leaks one log per job for the life of the process.

Always write a terminal state. The outcome is derived in SQL in a single UPDATE rather than split across a success branch and a failure branch. A job whose branch does not fire sits on running for ever while a client waits for a push that never comes. A runner killed mid-job cannot write anything at all, so the status route reports a job still running past the detached-run ceiling as timed_out rather than as progress.

RETURNING, not INSERT-then-SELECT. Reading the new row back with WHERE owner = ? ORDER BY created_at DESC LIMIT 1 looks equivalent and is not: two submits from the same owner in the same second both read the newest row, and one caller is handed the other's job id. SQLite 3.35+ supports RETURNING and the engine bundles a newer one.

Limits

Detached runs are capped per deployment and a request over the cap is refused with 429 rather than queued — a queue would only move the unboundedness somewhere less visible. A run that hangs is abandoned after 15 minutes so it cannot hold capacity indefinitely.

Configuration

jobs.yml

name: AsyncJobApi
description: >
  Expose your own long-running work as an async API. The caller POSTs, gets a job id back in
  milliseconds, and the answer arrives on a realtime channel — with a status route as the
  fallback for a client that missed the push or came back later. Nothing holds a request open,
  so work that outlives a proxy timeout still completes and is still retrievable.

docs: true

# ── The problem this solves ─────────────────────────────────────────────────────
# A request that takes 90 seconds does not fail because the work is hard. It fails because
# something between you and the caller stops waiting: Cloudflare cuts a proxied request at
# ~100s and returns a 524, most load balancers default to 60s, and a phone that changes
# network drops it sooner than either. The work usually FINISHES — the result just has
# nowhere to go, because the only place it existed was that connection.
#
# The fix is to stop making the connection the state:
#
#   1. POST /jobs/submit    -> a row is written, a job id comes straight back
#   2. the work runs DETACHED, with nobody waiting on it
#   3. on completion the answer is PUSHED to `jobs:<owner>` and written to the row
#   4. GET-style /jobs/status is the fallback for whoever missed step 3
#
# PUSH IS THE PRIMARY PATH, polling is the safety net. A client that holds the socket learns
# the moment the job lands; polling exists for the reload, the cold start, and the client that
# cannot hold a socket at all. Building it the other way round — poll first, push bolted on —
# gives you the latency of polling and the complexity of both.
#
# ── SELF-HOSTED ONLY ────────────────────────────────────────────────────────────
# `respond: early` is honoured by the self-hosted engine. Managed does not check the key yet:
# there the interface runs SYNCHRONOUSLY and the caller waits, which is the one thing this
# pack exists to avoid. Nothing errors and the config validates either way, so the symptom is
# just a slow request. Run this self-hosted until that changes.
#
# ── Setup ───────────────────────────────────────────────────────────────────────
# SQLite by default so it runs with no external database. Point DB at Postgres for anything
# real: SQLite is single-writer, and job rows are written from every detached run at once.
#   POST /jobs/init      once, to create the table
#   POST /jobs/submit    {"owner":"alice","payload":{...}}
#   POST /jobs/status    {"owner":"alice","job_id":"..."}

global:
  databases:
    main:
      driver: sqlite
      conn_string: "sqlite://async-jobs.db"

interfaces:

  # ── One-time setup ──────────────────────────────────────────────────────────────
  jobs-init:
    output: http
    method: POST
    summary: Create the jobs table
    description: Run once. Idempotent.
    tags: [setup, jobs]
    actions:
      - name: CreateTable
        database: main
        query: |-
          CREATE TABLE IF NOT EXISTS jobs (
            id          TEXT PRIMARY KEY,
            owner       TEXT NOT NULL,
            status      TEXT NOT NULL DEFAULT 'running',
            result      TEXT,
            error       TEXT,
            created_at  TEXT NOT NULL,
            finished_at TEXT
          );

  # ── The socket the caller watches ───────────────────────────────────────────────
  # Push is the primary delivery path. A caller subscribes to its own channel and is told the
  # moment its job lands, with no polling at all.
  live:
    ws: "on"
    subscribe_authorizer: authorize-owner
    summary: Realtime socket carrying job completions
    description: Subscribe to `jobs:<owner>` and receive each job's outcome as it finishes.
    tags: [websocket, realtime, jobs]
    actions:
      - name: Ping
        input: a|body|

  # Channel ACL. Bound to the owner in the token so one caller cannot watch another's jobs by
  # guessing a name. Swap the demo check for your real identity claim.
  authorize-owner:
    actions:
      - name: Verify
        input: a|headers|
        assert:
          http_code_on_error: 403
          error_message: "a valid token is required"
          tests:
            - value: airpipe-jwt
              is_valid_jwt: a|ap_var::JWT_SECRET|
      - name: Bind
        run_when_succeeded:
          actions: [Verify]
          http_code_on_error: 403
        input: a|body|
        assert:
          error_message: "channel not allowed for this owner"
          tests:
            - value: channel
              is_equal_to: "jobs:a|Verify::jwt_claims.owner|"

  # ── 1. Submit ───────────────────────────────────────────────────────────────────
  # Synchronous ON PURPOSE, and it is the reason this pack is three interfaces rather than one.
  # `respond: early` commits the status code BEFORE any action runs, so an interface carrying
  # it cannot reject a bad token — it would answer 202 and then fail silently. Authentication
  # therefore lives here, and only the work detaches.
  jobs-submit:
    output: http
    method: POST
    summary: Start a job and get its id immediately
    description: Returns {job_id, status} in milliseconds. The answer arrives on the channel.
    tags: [jobs, async]
    accepts:
      example:
        owner: alice
        payload: { report: "quarterly" }
    actions:
      - name: Auth
        input: a|headers|
        assert:
          http_code_on_error: 401
          error_message: "a valid token is required"
          tests:
            - value: airpipe-jwt
              is_valid_jwt: a|ap_var::JWT_SECRET|
        # is_valid_jwt decodes the token and exposes its claims as `jwt_claims`, which is where
        # `owner` comes from below. The whole point is that it is NOT read from the request.
        post_transforms:
          - extract_value: jwt_claims

      # The owner is taken from the VERIFIED TOKEN, never from the body. Reading it from the
      # request would let a caller file a job under someone else's name -- and then not receive
      # it either, since the channel ACL binds to the claim, so the push goes to the owner they
      # borrowed. Only `payload` comes from the caller.
      - name: CheckInput
        run_when_succeeded: [Auth]
        input: a|body|
        hide_data_on_success: true
        assert:
          http_code_on_error: 422
          error_message: "payload is required"
          tests:
            - value: payload
              is_not_null: true

      # The row exists BEFORE the work starts. That is what makes the outcome survive a closed
      # tab: the answer lives in a table, not in whoever happened to be connected.
      # RETURNING, not INSERT-then-SELECT. Reading the row back with
      # `WHERE owner = ? ORDER BY created_at DESC LIMIT 1` looks equivalent and is not: two
      # submits from the same owner in the same second both read the newest row, so one caller
      # is handed the other's job id and then polls a job that is not theirs. RETURNING hands
      # back the row that was just written, which is the only one that can be right.
      # SQLite 3.35+ supports it and the engine bundles a newer one.
      - name: CreateJob
        run_when_succeeded: [CheckInput]
        database: main
        query: |-
          INSERT INTO jobs (id, owner, status, created_at)
          VALUES (?, ?, 'running', strftime('%Y-%m-%dT%H:%M:%SZ','now'))
          RETURNING id, owner, status, created_at;
        params:
          - a|uuid|
          - a|Auth::owner|
        post_transforms:
          - extract_value: "[0]"

      # Starts the detached runner. It answers in milliseconds because it is `respond: early`,
      # so this call does not inherit the job's runtime.
      - name: FireRunner
        run_when_succeeded: [CreateJob]
        hide_data_on_success: true
        timeout: 15000
        http:
          expect_status: any
          url: "a|ap_var::SELF_URL|/jobs-run"
          method: POST
          headers:
            content-type: application/json
            airpipe-jwt: a|headers::airpipe-jwt|
          body: |
            {
              "job_id": a|CreateJob::id->double_quote|,
              "owner": a|Auth::owner->double_quote|,
              "payload": a|CheckInput::payload|
            }

      - name: Accepted
        run_when_succeeded: [CreateJob]
        input: a|CreateJob|
        post_transforms:
          - add_attribute:
              accepted: true
              watch: "jobs:a|Auth::owner|"
          - keep_attributes: ["accepted", "id", "status", "watch"]

  # ── 2. The detached runner ──────────────────────────────────────────────────────
  # `respond: early`: answers at once, finishes on the runtime. Detached runs are capped per
  # deployment and a request over the cap is REFUSED with 429 rather than queued — a queue here
  # would only move the unboundedness somewhere less visible.
  jobs-run:
    output: http
    method: POST
    respond: early
    summary: Runs the job with nobody waiting on it
    description: Internal. Does the work, records the outcome, pushes it to the owner's channel.
    tags: [jobs, async, internal]
    actions:
      - name: Auth
        input: a|headers|
        assert:
          tests:
            - value: airpipe-jwt
              is_valid_jwt: a|ap_var::JWT_SECRET|
        post_transforms:
          - extract_value: jwt_claims

      - name: CheckInput
        run_when_succeeded: [Auth]
        input: a|body|
        hide_data_on_success: true
        assert:
          tests:
            - value: job_id
              is_not_empty: true
            - value: owner
              is_not_empty: true

      # ── YOUR WORK GOES HERE ──────────────────────────────────────────────────────
      # Replace this with the thing that takes 90 seconds: the render, the export, the model
      # call, the report build. Everything around it stays as it is.
      - name: DoWork
        run_when_succeeded: [CheckInput]
        timeout: 900000
        http:
          expect_status: any
          url: "a|ap_var::WORK_URL|"
          method: POST
          idle_timeout: 60s
          headers:
            content-type: application/json
          body: a|CheckInput::payload|

      # ONE update, not a success branch and a failure branch. Two actions gated on the HTTP
      # status looks tidier and was the first shape here, but it depends on conditional-action
      # semantics being exactly what you assume, and a job whose branch does not fire sits on
      # `running` for ever while a client waits for a push that never comes. Deriving the
      # outcome in SQL always writes a terminal state, whatever the work returned.
      - name: RecordOutcome
        depends_on: [DoWork]
        database: main
        query: |-
          UPDATE jobs
             SET status      = CASE WHEN ? = 200 THEN 'done' ELSE 'failed' END,
                 result      = CASE WHEN ? = 200 THEN ?    ELSE NULL END,
                 error       = CASE WHEN ? = 200 THEN NULL ELSE 'work returned HTTP ' || ? END,
                 finished_at = strftime('%Y-%m-%dT%H:%M:%SZ','now')
           WHERE id = ? AND owner = ?;
        # `a|DoWork::body|`, NOT `a|DoWork::body->to_json|`. The modifier form does not resolve
        # here, and a param that fails to interpolate SKIPS the whole action -- reported as
        # "pre-assert did not pass", which points nowhere near the cause. The symptom was a job
        # stuck on `running` for ever while its completion push had already gone out.
        params:
          - a|DoWork::status->default(0)|
          - a|DoWork::status->default(0)|
          - a|DoWork::body|
          - a|DoWork::status->default(0)|
          - a|DoWork::status->default(0)|
          - a|CheckInput::job_id|
          - a|Auth::owner|

      - name: Notify
        depends_on: [RecordOutcome]
        ws_publish:
          channels: "jobs:a|Auth::owner|"
          # A mapping, not a `|` block: a block scalar publishes the JSON as a STRING and every
          # subscriber then has to parse `data` a second time. Measured on the wire before this
          # was written -- the frame carried "data":"{\n \"job_id\"...}" rather than an object.
          data:
            job_id: a|CheckInput::job_id|
            status: a|DoWork::status->default(0)|

  # ── 3. The fallback ─────────────────────────────────────────────────────────────
  # For the client that reloaded, was offline when the push went out, or cannot hold a socket.
  # Not the primary path — if you find yourself polling this on a timer, subscribe instead.
  jobs-status:
    output: http
    method: POST
    summary: Read a job's outcome
    description: Fallback for a client that missed the push.
    tags: [jobs, async]
    accepts:
      example:
        owner: alice
        job_id: "8f2a1c3e-..."
    actions:
      - name: Auth
        input: a|headers|
        assert:
          http_code_on_error: 401
          error_message: "a valid token is required"
          tests:
            - value: airpipe-jwt
              is_valid_jwt: a|ap_var::JWT_SECRET|
        post_transforms:
          - extract_value: jwt_claims

      # Same rule on the read side: the job is scoped to the token's owner, so a caller can only
      # ever read their own jobs and does not have to be trusted to say who they are.
      - name: CheckInput
        run_when_succeeded: [Auth]
        input: a|body|
        hide_data_on_success: true
        assert:
          http_code_on_error: 422
          error_message: "job_id is required"
          tests:
            - value: job_id
              is_not_empty: true

      # Scoped by owner as well as id, so one caller cannot read another's job by guessing.
      - name: ReadJob
        run_when_succeeded: [CheckInput]
        database: main
        # A runner killed mid-job -- a deploy, a restart, an OOM -- never gets to write a
        # terminal state, and the row would sit on `running` for ever with a client waiting on
        # a push that can no longer come. Anything still running past the detached-run ceiling
        # is reported as timed_out. The engine abandons a detached run at 15 minutes, so a job
        # older than 20 and still `running` is not slow, it is gone.
        query: |-
          SELECT id, owner,
                 CASE WHEN status = 'running'
                       AND created_at < strftime('%Y-%m-%dT%H:%M:%SZ','now','-20 minutes')
                      THEN 'timed_out' ELSE status END AS status,
                 result, error, created_at, finished_at
            FROM jobs
           WHERE id = ? AND owner = ?;
        params:
          - a|CheckInput::job_id|
          - a|Auth::owner|
        assert:
          http_code_on_error: 404
          error_message: "no such job for this owner"
          tests:
            - value: "[0]id"
              is_not_null: true
        post_transforms:
          - extract_value: "[0]"