Polling a long-running external job

HTTP_POLL is one task that does this. You give it the status URL and a condition that says "stop when this is true".

The shape

submit_job (HTTP)  ──>  await_job (HTTP_POLL)  ──>  SUCCEEDED ──> record artifact
                            │  polls the status URL          FAILED ──> TERMINATE
                            │  until terminationCondition
                            └─ sleeps between polls, holds nothing open

Why not a loop

A DO_WHILE wrapped around an HTTP task also works, and you will see it in older examples. It costs you more than it looks:

	DO_WHILE + HTTP	HTTP_POLL
Tasks in the execution	Two per iteration, forever growing	One
Backoff between polls	You build it	pollingStrategy
Poll ceiling	You count iterations yourself	maxPollCount
Reading the execution	Scroll past 40 iterations	One task with a poll count

The loop version also makes the interesting part — the termination condition — an expression buried in loopCondition, evaluated against loop state rather than the response.

The task

{
  "name": "await_job",
  "taskReferenceName": "await_job",
  "type": "HTTP_POLL",
  "inputParameters": {
    "http_request": {
      "uri": "${workflow.input.jobApiUrl}/jobs/${submit_job.output.response.body.jobId}",
      "method": "GET",
      "terminationCondition": "(function(){ var s = $.output.response.body.state; return s === 'SUCCEEDED' || s === 'FAILED'; })();",
      "pollingInterval": 60,
      "pollingStrategy": "FIXED",
      "maxPollCount": 60
    }
  }
}

HTTP_POLL takes the same http_request block as HTTP — uri, method, headers, body, accept, contentType, connectionTimeOut, readTimeOut, acceptedStatusCodes, outputFilter — plus four polling fields:

Field	Default	What it does
terminationCondition	—	Expression evaluated after each poll. Truthy stops the task
pollingInterval	—	Seconds between polls
pollingStrategy	—	FIXED, LINEAR_BACKOFF, or EXPONENTIAL_BACKOFF
maxPollCount	1000	Give up after this many polls

Writing the termination condition

The expression sees two objects:

• $.output — the current poll's result, including response.body, response.headers, response.statusCode
• $.input — the task's input

Return a boolean to say "done" or "keep going". You can also return a number for three-way control: 1 completes the task, 0 polls again, -1 fails it.

Terminate on failure too. A condition that only matches SUCCEEDED keeps polling a dead job until maxPollCount runs out. Match every terminal state and branch on the outcome afterwards:

(function(){ var s = $.output.response.body.state; return s === 'SUCCEEDED' || s === 'FAILED'; })();

Polling intervals have a server floor

pollingInterval is clamped to conductor.worker.http_poll.min_poll_interval, which defaults to 60 seconds. Asking for pollingInterval: 5 gets you 60 unless an operator lowered the floor. Size maxPollCount against the effective interval, not the one you asked for: 60 polls at 60 seconds is a one-hour ceiling.

Prerequisites

A running Conductor server and a job API to poll. A stub is included so you can run the shape without a vendor account.

Save this as job_stub_service.py and leave it running:

"""A stand-in for a slow third-party job API.

Run it before starting the workflow:

    python3 job_stub_service.py            # http://localhost:8089

Endpoints
    POST /jobs                 -> 202, returns {"jobId": "..."} and starts a job
    GET  /jobs/{id}            -> {"jobId","state","progress","result"}
                                  state goes QUEUED -> RUNNING -> SUCCEEDED
    POST /jobs/{id}/fail       -> force the job to FAILED on its next poll
    GET  /polls                -> how many times each job has been polled

The job advances one step per poll, so a workflow that polls it will see
QUEUED, then RUNNING, then SUCCEEDED, without any wall-clock waiting.
"""

import json
import uuid
from http.server import BaseHTTPRequestHandler, HTTPServer

JOBS = {}
POLLS = {}
STATES = ["QUEUED", "RUNNING", "RUNNING", "SUCCEEDED"]


class Handler(BaseHTTPRequestHandler):
    def _send(self, code, payload):
        body = json.dumps(payload).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self):
        path = self.path.split("?")[0]
        if path == "/jobs":
            job_id = f"job-{uuid.uuid4().hex[:8]}"
            JOBS[job_id] = {"step": 0, "failed": False}
            POLLS[job_id] = 0
            self._send(202, {"jobId": job_id, "state": "QUEUED"})
        elif path.endswith("/fail"):
            job_id = path.split("/")[2]
            if job_id in JOBS:
                JOBS[job_id]["failed"] = True
                self._send(200, {"jobId": job_id, "willFail": True})
            else:
                self._send(404, {"error": "no such job"})
        else:
            self._send(404, {"error": "not found"})

    def do_GET(self):
        path = self.path.split("?")[0]
        if path == "/polls":
            self._send(200, {"polls": POLLS})
            return
        if path.startswith("/jobs/"):
            job_id = path.split("/")[2]
            job = JOBS.get(job_id)
            if not job:
                self._send(404, {"error": "no such job"})
                return
            POLLS[job_id] = POLLS.get(job_id, 0) + 1
            if job["failed"]:
                self._send(200, {"jobId": job_id, "state": "FAILED",
                                 "progress": 100, "error": "upstream rejected the job"})
                return
            state = STATES[min(job["step"], len(STATES) - 1)]
            job["step"] += 1
            payload = {"jobId": job_id, "state": state,
                       "progress": min(100, job["step"] * 33)}
            if state == "SUCCEEDED":
                payload["result"] = {"rows": 4211, "artifact": f"s3://exports/{job_id}.csv"}
            self._send(200, payload)
            return
        self._send(404, {"error": "not found"})

    def log_message(self, *args):
        pass


if __name__ == "__main__":
    print("job stub listening on http://localhost:8089")
    HTTPServer(("127.0.0.1", 8089), Handler).serve_forever()

python3 job_stub_service.py       # http://localhost:8089

It advances one state per poll — QUEUED → RUNNING → RUNNING → SUCCEEDED — so you can watch the whole lifecycle without waiting on wall-clock time. POST /jobs/{id}/fail forces the failure branch, and GET /polls shows how many times each job was polled.

Runnable definition

Save this as http-poll-external-job.json:

{
  "name": "http_poll_external_job",
  "description": "Submit a job to a slow third-party API, then let a single HTTP_POLL task poll it until it reports a terminal state. No loop task, no worker.",
  "version": 1,
  "schemaVersion": 2,
  "timeoutSeconds": 3600,
  "timeoutPolicy": "TIME_OUT_WF",
  "inputParameters": [
    "jobApiUrl",
    "dataset"
  ],
  "tasks": [
    {
      "name": "submit_job",
      "taskReferenceName": "submit_job",
      "type": "HTTP",
      "inputParameters": {
        "http_request": {
          "uri": "${workflow.input.jobApiUrl}/jobs",
          "method": "POST",
          "body": { "dataset": "${workflow.input.dataset}" },
          "connectionTimeOut": 10000,
          "readTimeOut": 30000
        }
      }
    },
    {
      "name": "await_job",
      "taskReferenceName": "await_job",
      "type": "HTTP_POLL",
      "inputParameters": {
        "http_request": {
          "uri": "${workflow.input.jobApiUrl}/jobs/${submit_job.output.response.body.jobId}",
          "method": "GET",
          "connectionTimeOut": 10000,
          "readTimeOut": 30000,
          "terminationCondition": "(function(){ var s = $.output.response.body.state; return s === 'SUCCEEDED' || s === 'FAILED'; })();",
          "pollingInterval": 60,
          "pollingStrategy": "FIXED",
          "maxPollCount": 60
        }
      }
    },
    {
      "name": "route_on_job_state",
      "taskReferenceName": "route_job",
      "type": "SWITCH",
      "evaluatorType": "value-param",
      "expression": "state",
      "inputParameters": {
        "state": "${await_job.output.response.body.state}"
      },
      "decisionCases": {
        "SUCCEEDED": [
          {
            "name": "record_artifact",
            "taskReferenceName": "record_artifact",
            "type": "JSON_JQ_TRANSFORM",
            "inputParameters": {
              "body": "${await_job.output.response.body}",
              "queryExpression": "{jobId: .body.jobId, rows: (.body.result.rows // 0), artifact: (.body.result.artifact // \"\")}"
            }
          }
        ],
        "FAILED": [
          {
            "name": "terminate_job_failed",
            "taskReferenceName": "terminate_job_failed",
            "type": "TERMINATE",
            "inputParameters": {
              "terminationStatus": "FAILED",
              "workflowOutput": {
                "error": "remote_job_failed",
                "jobId": "${await_job.output.response.body.jobId}",
                "detail": "${await_job.output.response.body.error}"
              }
            }
          }
        ]
      },
      "defaultCase": []
    }
  ],
  "outputParameters": {
    "jobId": "${submit_job.output.response.body.jobId}",
    "finalState": "${await_job.output.response.body.state}",
    "rows": "${record_artifact.output.result.rows}",
    "artifact": "${record_artifact.output.result.artifact}"
  }
}

Register and run

conductor workflow create http-poll-external-job.json
conductor workflow start -w http_poll_external_job \
  -i '{"jobApiUrl":"http://localhost:8089","dataset":"orders_2026_q2"}'

Open Executions in the Conductor UI and select the new execution to review the task graph, and each task's inputs and outputs.

await_job stays as a single task and its poll count climbs. When the stub reports SUCCEEDED, the SWITCH records the artifact; force a failure with POST /jobs/{id}/fail and the same workflow terminates with remote_job_failed instead.

Cross-check what the vendor actually saw:

curl -s http://localhost:8089/polls

Production notes

• Match every terminal state in the condition, not just success, or a dead job polls until maxPollCount.
• pollingInterval has a server-side floor (min_poll_interval, default 60s). Your value is a request, not a guarantee.
• Set maxPollCount from a wall-clock budget. Interval × count is the real ceiling; give the workflow a timeoutSeconds above it.
• Use EXPONENTIAL_BACKOFF for jobs of unknown length so a five-hour job does not generate 300 identical requests.
• Poll a cheap endpoint. If the vendor's status call is rate-limited or returns the full payload, ask for a lightweight status URL, or use outputFilter to keep the response out of workflow state.
• The submit step needs an idempotency key. A retried submit that creates a second job leaves you polling the wrong one.
• Do not use it for sub-second work. Below the poll floor, a synchronous HTTP task is the right tool.

Related

• Wait and timer patterns — waiting on a signal or a clock rather than a status URL
• Task timeouts and retries — bounding the submit call
• Saga: compensating a partial failure — undoing a submitted job when a later step fails