Build a custom HTTP worker

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 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. From the server checkout, with the local stack running:

./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:

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.

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.

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.

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.

3. Pull one delivery

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

Handle these outcomes before reading a body:

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:

[[ "$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

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.

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

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:

[[ "$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 and 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. Dynamic appendedStages use string input and stage options; they are processed with the result. For the precise shapes, use the API reference.

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:

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.

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 and troubleshooting for operational diagnosis.