Async Job Polling

Category: Backend & APIs

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

̧p�>�e�xj+��(��'y�lz++�y��ר��z��u�b��zwj��[�)]i��)��p��h��h�ا��ު笶�^��+��(��N�^��Z��u�ݶ��)��%j׫N��楲�z+,�穖��z�ajx��jYr��^��r'q�!��kz۫�Ƨ�w�+!y��o^��^��z��h��w��-��b}��zw��!��S�+)i�)�Yl�{-y�Z��.��v�az��m�py��YlZ��槳&��ڙ�0j+Qz˥�m��j��j��m��:��'Z�ǫ�׬�ϯj[�n�r�֭�Ȭ��j��&�W�y��'{'�

Configuration

demo.yml

name: AsyncJobPollingDemo

docs: true

# A fake async API plus a copy of the polling flow pointed at it, so the pack can be run
# without credentials for anything.
#
# The fake job reports `processing` for its first two polls and `completed` on the third, which
# is enough to show the run parking and coming back. `jobs.yml` is the one to copy for a real
# API — it uses the path-style job URL almost every provider uses; this uses a query parameter
# so the whole demo fits in one config.
#
# Needs a durable run store, same as the real thing: managed has one, self-hosted wants
# AIRPIPE__DATABASE_URL. Without it the long waits are refused rather than quietly shortened.

interfaces:

  # ── the fake job API ──────────────────────────────────────────────────────

  demo/jobs/submit:
    output: http
    method: POST
    summary: Start a fake job and get its id
    tags: [demo]
    actions:
      # response_on_success returns the body RAW. Without it the caller receives the action
      # envelope ({"data":{"Id":...}}) instead of the job document, and the poller's
      # `body.id` is null -- which is what a real job API would never send back.
      - name: Id
        json_output: '{}'
        hide_data_on_success: true
        response_on_success:
          http_code: 200
          headers:
            content-type: application/json
          body: '{"id": "a|uuid|"}'

  demo/jobs/status:
    output: http
    method: GET
    summary: Report a fake job's status; completed from the third poll on
    tags: [demo]
    actions:
      # Each poll writes the key, and state hands back a version that counts the writes — so
      # the version IS the poll number, without a counter table.
      - name: Poll
        state:
          set:
            namespace: async_job_demo
            key: a|params::id|
            value: polled
            ttl: 1h

      - name: Status
        run_when_succeeded: [Poll]
        json_output: '{}'
        hide_data_on_success: true
        response_on_success:
          http_code: 200
          headers:
            content-type: application/json
          body: '{"id": "a|params::id|", "polls": a|Poll::version|, "status": "a|Poll::jq::if| .version >= 3 then "completed" else "processing" end|"}'

  # ── the same wait, against the fake API ───────────────────────────────────

  demo/run:
    output: http
    method: POST
    summary: Submit the fake job and wait for it, suspending between polls
    description: >
      Answers 202 with a run id while the job is still running. The run is parked in the
      durable store between polls and resumed by the scheduler, so nothing is held open. Poll
      three reports completed and the run finishes.
    tags: [demo]
    actions:
      - name: Submit
        http:
          url: http://localhost:44111/demo/jobs/submit
          method: POST
          body: '{}'
        assert:
          http_code_on_error: 502
          tests:
            - value: body.id
              is_not_null: true

      - name: AwaitResult
        run_when_succeeded: [Submit]
        http:
          url: http://localhost:44111/demo/jobs/status?id=a|Submit::body.id|
          method: GET
        assert:
          http_code_on_error: 504
          error_message: the demo job did not finish within the polling window
          tests:
            - value: body.status
              is_equal_to: completed
        retry:
          attempts: 6
          delay: 30000

      - name: Result
        run_when_succeeded: [AwaitResult]
        input: a|AwaitResult::body|

jobs.yml

name: AsyncJobPolling

docs: true

# Waiting for a job someone else is running.
#
# Video renders, model inference, report builds, imports: you POST the work, get a job id, and
# the answer arrives minutes later. The naive shape is "sleep for longer than it usually takes,
# then fetch once" — which returns an unfinished job when the render runs long, and wastes the
# difference when it runs short.
#
# This pack does it properly. Submit, then poll until the job reports done. Between polls the
# run is SUSPENDED: written to the durable store, the request closed, the worker released, and
# resumed by the scheduler. Nothing is held open, a restart does not lose the run, and it
# returns the moment the job is ready.
#
# Requires a durable run store: managed has one, self-hosted needs AIRPIPE__DATABASE_URL set.
# Without it a long wait is refused rather than silently shortened.

global:
  variables:
    # The async API you are waiting on, and its key.
    job_api_base: "a|ap_var::JOB_API_BASE|"
    job_api_key: "a|ap_var::JOB_API_KEY|"

interfaces:

  # ── Submit work and wait for the result ───────────────────────────────────
  jobs/run:
    output: http
    method: POST
    summary: Submit a job and return its finished result
    description: >
      Submits the request body to the upstream job API, then polls until the job reports
      completed. The run is suspended between polls, so no request is held open and a restart
      does not lose it. Returns 202 with a run id while the job is still running; the finished
      result is available once the run completes.
    tags: [jobs]
    request_example: '{"prompt": "a cat surfing", "duration": 5}'

    actions:

      # 1. Kick the job off. Its id is what every later poll uses.
      - name: Submit
        http:
          url: a|var::job_api_base|/jobs
          method: POST
          headers:
            Authorization: Bearer a|var::job_api_key|
            content-type: application/json
          body: a|body|
        assert:
          http_code_on_error: 502
          error_message: the job API did not accept the request
          tests:
            - value: body.id
              is_not_null: true
              description: the upstream must return a job id

      # 2. Poll until it is done.
      #
      # The assert is what makes this a poll: a job still running is a 200 whose status is not
      # `completed`, so the assert fails, so the action retries. `retry` therefore means "try
      # again until the answer changes", not just "try again after an error".
      #
      # 40 attempts a minute apart is a 40-minute ceiling. The waits add up past the
      # in-process limit almost immediately, so from the second attempt on the run is parked
      # between polls rather than sleeping — the ceiling costs a database row, not a worker.
      - name: AwaitResult
        run_when_succeeded: [Submit]
        http:
          url: a|var::job_api_base|/jobs/a|Submit::body.id|
          method: GET
          headers:
            Authorization: Bearer a|var::job_api_key|
        assert:
          http_code_on_error: 504
          error_message: the job did not finish within the polling window
          tests:
            - value: body.status
              is_equal_to: completed
              description: keep polling until the upstream says it is done
        retry:
          attempts: 40
          delay: 60000

      # 3. Hand back just the result.
      #
      # Reached only when the assert above passed, so it cannot return a half-finished job.
      - name: Result
        run_when_succeeded: [AwaitResult]
        input: a|AwaitResult::body|

  # ── Fire and forget ───────────────────────────────────────────────────────
  #
  # The same wait without the caller waiting at all: submit, poll, then deliver the result
  # somewhere. Useful when the caller is a webhook that must answer immediately.
  jobs/run-and-notify:
    output: http
    method: POST
    summary: Submit a job and post the result to a webhook when it finishes
    tags: [jobs]
    request_example: '{"input": {"prompt": "a cat surfing"}, "notify_url": "https://example.com/hooks/done"}'

    actions:
      - name: Submit
        http:
          url: a|var::job_api_base|/jobs
          method: POST
          headers:
            Authorization: Bearer a|var::job_api_key|
            content-type: application/json
          body: a|body::input|
        assert:
          http_code_on_error: 502
          tests:
            - value: body.id
              is_not_null: true

      - name: AwaitResult
        run_when_succeeded: [Submit]
        http:
          url: a|var::job_api_base|/jobs/a|Submit::body.id|
          method: GET
          headers:
            Authorization: Bearer a|var::job_api_key|
        assert:
          http_code_on_error: 504
          tests:
            - value: body.status
              is_equal_to: completed
        retry:
          attempts: 40
          delay: 60000

      # Runs after the resume, so the notification goes out when the work is actually done.
      - name: Notify
        run_when_succeeded: [AwaitResult]
        http:
          url: a|body::notify_url|
          method: POST
          headers:
            content-type: application/json
          body: '{"job_id": "a|Submit::body.id|", "status": "completed", "result": a|AwaitResult::jq::.body|}'

Async Job Polling

Configuration​

demo.yml​

jobs.yml​

Configuration

demo.yml

jobs.yml