---
title: "Build a custom HTTP worker"
description: "Bootstrap a worker, pull jobs, acquire and renew execution leases, submit results, and acknowledge deliveries over HTTP."
category: "Reference"
---

A custom worker can execute Pipelogiq stages using the external HTTP API without opening a RabbitMQ connection. Use this protocol when your language or network environment makes a direct broker client inconvenient. The Pipelogiq server still requires RabbitMQ and its orchestration worker: HTTP changes how your application receives jobs and reports results.

This guide describes the current `0.5.0` source. Follow [installation](/docs/installation) first, or use an existing deployment with an application API key. The examples use Bash, curl, and jq against `http://localhost:8081`. Use your deployment's HTTPS address outside a local environment.

## Prove the whole path first

The repository includes an [HTTP worker smoke script](https://github.com/pipelogiq/pipelogiq/blob/main/scripts/http-worker-smoke.sh). From the server checkout, with the local stack running:

```bash
./scripts/http-worker-smoke.sh
```

The script uses `BOOTSTRAP_API_KEY` from your shell or the root `.env`. To use another application key without putting it in shell history:

```bash
read -r -s -p 'Application API key: ' BOOTSTRAP_API_KEY
printf '\n'
export BOOTSTRAP_API_KEY
PIPELOGIQ_EXTERNAL_URL=http://localhost:8081 ./scripts/http-worker-smoke.sh
unset BOOTSTRAP_API_KEY
```

Expected result: a two-stage pipeline completes, the first stage records one retry, and the script prints `PASS`. It creates a worker record and a test pipeline. This is a short execution smoke test; it does not exercise long-running lease renewal or production load balancing.

## Credentials and endpoint requirements

An application API key and a worker session token are separate credentials. Dashboard JWTs are not worker credentials. Use the two explicit headers below; a single `Authorization: Bearer` value cannot represent both identities.

| Endpoint | Credential checked by the current server |
| --- | --- |
| `POST /workers/bootstrap` | `X-API-Key` |
| `POST /jobs/pull` | `X-API-Key`; the requested queue must belong to that application |
| `POST /stages/{id}/lease/acquire` and `/lease/renew` | `X-Worker-Session`, matched to the body `workerId` |
| `POST /stages/{id}/result` | Both `X-API-Key` and `X-Worker-Session`, for the same application |
| `POST /workers/heartbeat`, `/workers/events`, `/workers/shutdown` | `X-Worker-Session`, matched to the body `workerId` |
| `POST /jobs/ack` | The opaque delivery `token` in the body; no API-key or session check |

Treat delivery tokens as secrets too. Do not record keys, session tokens, bootstrap broker connection strings, or sensitive job payloads in logs. Sending both headers on worker requests is convenient and supported, but does not change which credential each handler checks.

## 1. Bootstrap your process

The following helper keeps response bodies and status codes separate. Run these examples in one Bash session. It sends an application key and, after bootstrap, the worker session header.

```bash
PIPELOGIQ_EXTERNAL_URL=http://localhost:8081
read -r -s -p 'Application API key: ' PIPELOGIQ_API_KEY
printf '\n'
SESSION=''
HANDLER=CustomHttpHandler
INSTANCE_ID="custom-http-$(date +%s)-$$"

http_post() {
  local path="$1" body="$2" response
  local headers=(-H "X-API-Key: $PIPELOGIQ_API_KEY")
  if [[ -n "$SESSION" ]]; then
    headers+=(-H "X-Worker-Session: $SESSION")
  fi
  response=$(curl -sS -w $'\n%{http_code}' \
    -X POST "$PIPELOGIQ_EXTERNAL_URL$path" \
    "${headers[@]}" -H 'Content-Type: application/json' \
    --data-binary "$body") || return 1
  HTTP_STATUS="${response##*$'\n'}"
  HTTP_BODY="${response%$'\n'*}"
}

http_post /workers/bootstrap "$(jq -nc \
  --arg instance "$INSTANCE_ID" --arg handler "$HANDLER" \
  '{workerName:"custom-http",instanceId:$instance,supportedHandlers:[$handler]}')"
[[ "$HTTP_STATUS" == 200 ]] || { printf 'Bootstrap failed: HTTP %s\n' "$HTTP_STATUS"; exit 1; }
WORKER_ID=$(jq -er .workerId <<<"$HTTP_BODY")
SESSION=$(jq -er .workerSessionToken <<<"$HTTP_BODY")
QUEUE_APP_ID=$(jq -er .application.appId <<<"$HTTP_BODY")
QUEUE_PATTERN=$(jq -er .queues.stageNextPattern <<<"$HTTP_BODY")
HEARTBEAT_SECONDS=$(jq -er .heartbeat.intervalSec <<<"$HTTP_BODY")
[[ "$QUEUE_PATTERN" == '{appId}_{handler}_StageNext' ]] || exit 1
QUEUE="${QUEUE_APP_ID}_${HANDLER}_StageNext"
```

`workerName` is required. Optional bootstrap fields include `instanceId`, `workerVersion`, `sdkVersion`, `environment`, `hostName`, `pid`, `supportedHandlers`, `capabilities`, and `metadata`. Give concurrently running processes different instance IDs. Bootstrapping the same application and instance again reuses its worker record and replaces its session token.

The response includes worker identity, application identity, queue names, heartbeat timing, broker configuration, and observability links. HTTP workers ignore the broker connection string. `application.applicationId` is numeric; `application.appId` is a queue prefix such as `app_12`. A handler named `CustomHttpHandler` therefore consumes `app_12_CustomHttpHandler_StageNext`.

Sessions expire after `WORKER_SESSION_TTL`, which defaults to 24 hours. Heartbeats do not extend that expiry. Plan controlled re-bootstrap before expiry, and distinguish an expired session from an invalid application key when handling `401`.

## 2. Announce readiness and create a test job

Send a heartbeat once your worker can accept jobs, then continue at the returned interval, normally 15 seconds. A production implementation runs heartbeats independently of pulling and executing jobs.

```bash
http_post /workers/heartbeat "$(jq -nc --arg worker "$WORKER_ID" \
  '{workerId:$worker,state:"ready",inFlightJobs:0,jobsProcessed:0,jobsFailed:0}')"
[[ "$HTTP_STATUS" == 200 ]] || exit 1

http_post /pipelines/idempotent "$(jq -nc \
  --arg key "$INSTANCE_ID" --arg handler "$HANDLER" \
  '{idempotencyKey:$key,name:"custom-http-example",stages:[{
    stageName:"hello",stageHandlerName:$handler,input:"{\"message\":\"hello\"}",
    options:{maxRetries:2,retryInterval:5}
  }]}')"
[[ "$HTTP_STATUS" == 201 || "$HTTP_STATUS" == 200 ]] || exit 1
PIPELINE_ID=$(jq -er .pipeline.id <<<"$HTTP_BODY")
```

Supported heartbeat states are `starting`, `ready`, `degraded`, `draining`, `stopped`, `error`, and `offline`. An omitted or unrecognized state keeps the previous state. Optional counters and metrics include uptime, in-flight jobs, processed and failed jobs, CPU, and memory. An HTTP worker need not claim a direct broker connection: omit `brokerConnected` if it is not meaningful.

A worker is considered offline after the configured silence interval, normally 45 seconds. Worker liveness also matters to automatic schedules; see [schedules](/docs/schedules).

## 3. Pull one delivery

```bash
http_post /jobs/pull "$(jq -nc --arg queue "$QUEUE" \
  '{queue:$queue,waitSeconds:20}')"
```

Handle these outcomes before reading a body:

| Status | Action |
| --- | --- |
| `200` | Decode one delivery and acquire its lease. |
| `204` | The queue was empty until the deadline. Start another long poll. |
| `400` | Fix the queue or `waitSeconds`; the accepted range is 0–20. |
| `401` | Resolve the application API key. Pull does not validate a worker session. |
| `403` | Fix the application/handler queue name. |
| `429` | Back off before pulling again; the API has reached its in-flight delivery limit. |
| `500` or transport failure | Retry with backoff; investigate the API and broker if persistent. |

`waitSeconds` defaults to zero, which returns immediately for an empty queue. Positive values use server polling at roughly 250 ms intervals. Configure your HTTP client's timeout above the requested long-poll duration.

When the status is `200`, retain both the delivery and its opaque token:

```bash
[[ "$HTTP_STATUS" == 200 ]] || exit 1
DELIVERY="$HTTP_BODY"
TOKEN=$(jq -er .token <<<"$DELIVERY")
STAGE_ID=$(jq -er .payload.stageId <<<"$DELIVERY")
EXECUTION_ID=$(jq -er .payload.executionId <<<"$DELIVERY")
ATTEMPT=$(jq -er .payload.attempt <<<"$DELIVERY")
```

The `payload` contains `stageId`, `pipelineId`, `executionId`, `attempt`, handler name, string `input`, string `prevStageOutput`, and context items. It may also contain the pipeline idempotency key, timeout seconds, and tracing fields. Parse input according to your application's contract. Preserve the execution identity unchanged.

## 4. Acquire before executing; renew while running

```bash
LEASE_BODY=$(jq -nc --arg execution "$EXECUTION_ID" --arg worker "$WORKER_ID" \
  '{executionId:$execution,workerId:$worker}')
http_post "/stages/$STAGE_ID/lease/acquire" "$LEASE_BODY"
[[ "$HTTP_STATUS" == 200 ]] || exit 1
LEASE_ACQUIRED=$(jq -r .acquired <<<"$HTTP_BODY")
```

Only `acquired:true` authorizes this worker to execute the delivery. The response supplies the accepted attempt and `leaseExpiresAt`. Acquiring a lease is not reentrant: acquiring again while the same execution has an active lease can return `lease_held`.

| Refusal reason | Meaning and response |
| --- | --- |
| `stale_execution`, `stage_not_active`, `pipeline_terminal` | This delivery must not run. Acknowledge and discard it. |
| `lease_held`, `lease_not_acquired` | Another acquisition owns or won the execution. Do not run this delivery; acknowledge the duplicate. |
| `lease_expired` | A running execution needs server lease recovery. Do not execute it or spin on acquisition; recovery is responsible for redispatch. |
| `application_mismatch` | Worker and stage belong to different applications. Stop this consumer and correct its configuration. |

The current lease duration is 60 seconds. Use `leaseExpiresAt` to schedule renewal with margin, typically around every 30 seconds:

```bash
http_post "/stages/$STAGE_ID/lease/renew" "$LEASE_BODY"
```

Renewal must return `200` with `acquired:true`. A definitive refusal means stop the handler cooperatively and do not start further side effects. If the network is unavailable, do not continue past your last confirmed lease expiry. Pass cancellation and timeout signals to your own HTTP calls and database work. Pipelogiq cannot forcibly stop arbitrary user code.

**The execution lease and delivery token have separate lifetimes.** Renewing a lease does not extend the pulled delivery's visibility timeout. `GATEWAY_VISIBILITY_TIMEOUT` defaults to 60 seconds; an expiry sweep attempts to requeue overdue deliveries. Configure it above normal job duration plus result-posting margin, or expect redeliveries and rejected duplicate acquisitions. There is no HTTP endpoint to renew a delivery token.

## 5. Execute and submit the result

For this example, execute only after acquisition succeeds, then submit a successful result:

```bash
[[ "$LEASE_ACQUIRED" == true ]] || exit 1
RESULT_BODY=$(jq -nc \
  --argjson pipeline "$PIPELINE_ID" --argjson stage "$STAGE_ID" \
  --arg execution "$EXECUTION_ID" --argjson attempt "$ATTEMPT" \
  '{pipelineId:$pipeline,stageId:$stage,executionId:$execution,attempt:$attempt,
    isSuccess:true,result:"{\"ok\":true}"}')
http_post "/stages/$STAGE_ID/result" "$RESULT_BODY"
[[ "$HTTP_STATUS" == 202 ]] || exit 1
```

Always send `isSuccess` explicitly: omission decodes as `false`. `stageId` must match the URL; `executionId` must be nonempty and `attempt` positive. `result` is a string capped at 262,144 bytes. The complete JSON request is capped at 327,680 bytes, including logs, context, and appended stages. Unknown fields are tolerated for SDK compatibility; accepting them does not mean they affect execution.

For failure, send `isSuccess:false`, a useful `errorCode`, and an explanation in `result`. `retryable:false` disables automatic retries. `retryable:true` still requires an applicable retry policy or stage retry options and cannot override terminal error codes such as `BUSINESS_REJECTED` or `VALIDATION_ERROR`. See [reliability](/docs/reliability) and [policies](/docs/policies).

Results may include `logs` with `message`, `logLevel`, and an RFC 3339 `created` time; context updates with `key`, string `value`, `valueType`, and `isSensitive`; or tombstones such as `{"key":"temporary","isDeleted":true}`. `isWaitingForApproval:true` parks the stage for the explicit [approval/resume flow](/docs/concepts). Dynamic `appendedStages` use string input and stage options; they are processed with the result. For the precise shapes, use the [API reference](/docs/api).

A `202` response confirms persistent, publisher-confirmed acceptance by RabbitMQ. It does not confirm a database state change. The orchestration worker consumes the result asynchronously and fences stale or duplicate executions. A stale result can receive `202` and later have no effect. Read pipeline status to verify the final outcome.

On `503` or an unknown network outcome, retain the computed result and retry its submission with backoff and the same execution metadata; do not run the business action again just to reconstruct a response. If ownership is lost, let recovery decide the next execution. Design external side effects with application-level idempotency because lease recovery cannot make them exactly once.

## 6. Acknowledge, then inspect the pipeline

After `202`, acknowledge the original delivery token:

```bash
http_post /jobs/ack "$(jq -nc --arg token "$TOKEN" \
  '{token:$token,requeue:false}')"
printf 'Acknowledgement HTTP status: %s\n' "$HTTP_STATUS"

curl -fsS "$PIPELOGIQ_EXTERNAL_URL/pipelines/$PIPELINE_ID" \
  -H "X-API-Key: $PIPELOGIQ_API_KEY" | jq '{id,status,stages}'
```

Expected result: acknowledgement returns `200`; the pipeline eventually becomes `Completed`. Repeat the status read if result consumption is still in progress.

Acknowledge without executing when discarding a stale or duplicate delivery as described above. Requeue with `requeue:true` when intentionally returning unstarted work, such as during shutdown. Requeueing changes broker delivery state; it does not reset a running execution lease or itself create a new execution attempt.

`404` from acknowledgement means that API process does not hold the token. It can be expired, already acknowledged, or held by a different API replica. It does not prove the message was successfully requeued. Pending delivery handles live in the API process, so route pull and acknowledgement to the same instance. After an ambiguous acknowledgement, check the pipeline outcome and allow normal delivery/lease recovery; avoid repeating side effects.

## 7. Send lifecycle events and stop gracefully

Worker events appear in the worker activity feed. Submit a nonempty batch; the default maximum is 200 events per request, configurable with `WORKER_EVENTS_MAX_BATCH`.

```bash
http_post /workers/events "$(jq -nc --arg worker "$WORKER_ID" \
  '{workerId:$worker,events:[{level:"INFO",eventType:"worker.example_completed",
    message:"Custom HTTP example completed"}]}')"

http_post /workers/shutdown "$(jq -nc --arg worker "$WORKER_ID" \
  '{workerId:$worker,reason:"example complete"}')"
unset SESSION PIPELOGIQ_API_KEY TOKEN DELIVERY HTTP_BODY
```

Production shutdown should first stop new pulls, report `draining`, and finish or cooperatively stop in-flight work while maintaining heartbeats and leases. Report accepted results before acknowledgement. Finally call shutdown to mark the worker stopped. This endpoint records lifecycle state; it is not a session-token revocation operation.

Use stage-result logs and worker events for this integration. The legacy `/logs` route has different, optional body-key behavior and should not be treated as the authenticated worker logging endpoint. See [observability](/docs/observability) and [troubleshooting](/docs/troubleshooting) for operational diagnosis.