Async Job API (Postgres)

Category: Engineering & DevOps

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
`DATABASE_URL`	`postgres://user:pass@host:5432/db`

Postgres — the shape to deploy. The SQLite sibling (Async Job API) exists to be run with nothing installed; use it to see the pattern, use this one to run it. 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.

What Postgres buys beyond not blocking:

result jsonb — the outcome is queryable, not an opaque blob: SELECT result->>'rows' FROM jobs WHERE status = 'done'
an index on (owner, created_at DESC) for listing a caller's jobs
now() - interval instead of string-compared timestamps
elapsed_secs computed in the status response

# 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 = $1 ORDER BY created_at DESC LIMIT 1 looks equivalent and is not: two submits from the same owner in the same moment both read the newest row, and one caller is handed the other's job id.

multi: true on the schema block. The native driver prepares every query and a prepared statement holds one command, so a CREATE TABLE plus CREATE INDEX in one query fails with 42601 without it.

Retention

Rows accumulate with their full result payload. Decide a retention window before this table gets large:

DELETE FROM jobs WHERE finished_at < now() - interval '30 days';

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: AsyncJobApiPostgres
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 ───────────────────────────────────────────────────────────────────────
# Postgres, for a deployment that runs more than one job at a time. The SQLite sibling
# (`async-job-api`) exists to be run with nothing installed; this one is the shape to deploy.
# 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.
#
# What Postgres buys here beyond not blocking:
#   * `result jsonb` -- the outcome is queryable (`result->>'rows'`), not an opaque string
#   * a real index on (owner, created_at) for listing a caller's jobs
#   * `now() - interval` instead of string-compared timestamps
#
#   AIRPIPE__DATABASE_URL   postgres://user:pass@host:5432/db
#   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: postgres
      conn_string: a|ap_var::DATABASE_URL|

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
        # Two statements in one query. The native driver prepares every query and a prepared
        # statement holds only one command, so this fails with 42601 without `multi`.
        multi: true
        query: |-
          CREATE TABLE IF NOT EXISTS jobs (
            id          uuid PRIMARY KEY DEFAULT gen_random_uuid(),
            owner       text NOT NULL,
            status      text NOT NULL DEFAULT 'running',
            result      jsonb,
            error       text,
            created_at  timestamptz NOT NULL DEFAULT now(),
            finished_at timestamptz
          );
          -- Listing a caller's jobs is the query this table exists to serve second.
          CREATE INDEX IF NOT EXISTS jobs_owner_created_idx ON jobs (owner, created_at DESC);

  # ── 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.
      - name: CreateJob
        run_when_succeeded: [CheckInput]
        database: main
        query: |-
          INSERT INTO jobs (owner, status) VALUES ($1, 'running')
          RETURNING id, owner, status, created_at;
        params:
          - 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 $1::int = 200 THEN 'done' ELSE 'failed' END,
                 result      = CASE WHEN $1::int = 200 THEN $2::jsonb ELSE NULL END,
                 error       = CASE WHEN $1::int = 200 THEN NULL
                                    ELSE 'work returned HTTP ' || $1::text END,
                 finished_at = now()
           WHERE id = $3::uuid AND owner = $4;
        # `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::body|
          - 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 < now() - interval '20 minutes'
                      THEN 'timed_out' ELSE status END AS status,
                 result, error, created_at, finished_at,
                 EXTRACT(EPOCH FROM (COALESCE(finished_at, now()) - created_at))::int AS elapsed_secs
            FROM jobs
           WHERE id = $1::uuid AND owner = $2;
        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]"

Async Job API (Postgres)

The problem​

Push first, poll as the fallback​

What's included​

Setup​

The token​

Watching the channel​

Three things worth copying​

Retention​

Limits​

Configuration​

jobs.yml​