Guide for processing jobs using the Spooled Worker runtime.
from spooled import SpooledClient
from spooled.worker import SpooledWorker
# Create client
client = SpooledClient(api_key="sp_live_...")
# Create worker
worker = SpooledWorker(
client,
queue_name="my-queue",
concurrency=10,
)
# Register job handler
@worker.process
def handle_job(ctx):
print(f"Processing: {ctx.job_id}")
# Your job logic here
result = do_work(ctx.payload)
return {"success": True, "result": result}
# Start worker (blocking)
worker.start()worker = SpooledWorker(
client,
queue_name="my-queue",
concurrency=10, # Max concurrent jobs
poll_interval=1.0, # Seconds between polls
lease_duration=30, # Job lease in seconds
heartbeat_fraction=0.5, # Heartbeat at 50% of lease
shutdown_timeout=30.0, # Graceful shutdown timeout
hostname="worker-1", # Custom hostname
worker_id="worker-1", # Stable id: registration becomes an upsert
worker_type="python", # Worker type identifier
# version defaults to spooled.__version__
metadata={"env": "prod"}, # Custom metadata
)worker_id is optional, and what it changes is what happens when a worker restarts.
Pass a stable id (1-128 characters from A-Z a-z 0-9 . _ -) and registration becomes an upsert: the worker reuses the same row every time it starts, and re-registering an id you already own is not charged against the plan worker cap.
Omit it and the server mints a fresh UUID on every registration. The row from the previous run keeps occupying the plan worker cap until the stale-worker reaper clears it, which takes about two minutes — so a worker that crash-loops on a tight plan can register itself into a RateLimitError with code QUOTA_EXCEEDED. Give any long-lived or supervised worker a stable id; the UUID default is fine for one-off and short-lived workers.
An id that already belongs to a different organization is rejected with a 409 ConflictError, so derive ids from something you control, such as the pod or machine name:
import os
worker = SpooledWorker(
client,
queue_name="my-queue",
worker_id=os.environ.get("HOSTNAME", "worker-local"),
)The job handler receives a context object:
@worker.process
def handle_job(ctx):
# Job information
ctx.job_id # Job ID
ctx.queue_name # Queue name
ctx.payload # Job payload (dict)
ctx.retry_count # Current retry count
ctx.max_retries # Maximum retries allowed
# Abort signal (threading.Event). It is set when the job exceeds its
# timeout_seconds or the worker's shutdown_timeout runs out - not as soon as
# stop() is called. Long handlers should check it between steps.
if ctx.signal.is_set():
print("Job was aborted!")
return None
# Logging
ctx.log("info", "Processing started", {"key": "value"})
# Progress is recorded as a local job log entry (percent clamped to 0-100);
# it is not persisted server-side
ctx.progress(50, "Halfway done")
return {"result": "done"}Register handlers for worker events:
@worker.on("started")
def on_started(event):
print(f"Worker {event.worker_id} started on {event.queue_name}")
@worker.on("stopped")
def on_stopped(event):
print(f"Worker stopped: {event.reason}")
@worker.on("error")
def on_error(event):
print(f"Worker error: {event.error}")
@worker.on("job:claimed")
def on_claimed(event):
print(f"Job {event.job_id} claimed")
@worker.on("job:started")
def on_started(event):
print(f"Job {event.job_id} started")
@worker.on("job:completed")
def on_completed(event):
print(f"Job {event.job_id} completed: {event.result}")
@worker.on("job:failed")
def on_failed(event):
print(f"Job {event.job_id} failed: {event.error}")
print(f"Will retry: {event.will_retry}")import signal
import sys
def shutdown(signum, frame):
print("Shutting down...")
worker.stop() # Drains running jobs (up to shutdown_timeout), then deregisters
client.close()
sys.exit(0)
signal.signal(signal.SIGTERM, shutdown)
signal.signal(signal.SIGINT, shutdown)
worker.start()stop() (sync) / await worker.stop() (async) drains the worker:
- It stops claiming new jobs. A claim that was already in flight is left unstarted; deregistering re-queues it for another worker.
- Running handlers keep going for up to
shutdown_timeoutseconds. Their job heartbeats and the worker heartbeat keep running, so the leases stay valid.ctx.signalis not set yet. - A handler that returns in that window is acked with
complete, one that raises withfail- exactly as if no shutdown had been requested. A result is never dropped. - Handlers still running after
shutdown_timeoutgetctx.signalset (the async worker also cancels the task) and their jobs are failed with "Worker shutdown timeout", so they retry elsewhere. A sync handler thread cannot be killed: it keeps running but its late result is discarded. - The worker deregisters last, because deregistering re-queues every job the worker still holds.
start() returns only after stop() has finished, so it is safe to close the client
right after start() returns, even when stop() ran on another thread. Set
shutdown_timeout at least as long as your longest handler.
A job's timeout_seconds (set when it is created, 300 by default) is enforced by the
worker. When a handler runs longer, the worker sets ctx.signal, fails the job with
"Job timeout exceeded" and stops extending its lease; the server then retries the job
or dead-letters it like any other failure. The async handler is also cancelled. A sync
handler thread cannot be killed, so it keeps occupying one of the concurrency slots
until it returns - give blocking calls their own timeouts, or check ctx.signal.
Exceptions in job handlers are automatically caught and reported:
@worker.process
def handle_job(ctx):
if ctx.payload.get("should_fail"):
raise ValueError("Intentional failure")
return {"success": True}The job will be marked as failed and retried if retries remain. The failure text is
str(exception): an empty message is sent as "Job failed" and anything longer than
2048 characters is truncated to 2048, because the API rejects both.
import asyncio
from spooled import AsyncSpooledClient
from spooled.worker import AsyncSpooledWorker
async def main():
async with AsyncSpooledClient(api_key="sp_live_...") as client:
worker = AsyncSpooledWorker(
client,
queue_name="my-queue",
concurrency=10,
worker_id="worker-1",
)
@worker.process
async def handle_job(ctx):
# Async job processing
await asyncio.sleep(1)
return {"success": True}
await worker.start()
asyncio.run(main())Each claimed job is tracked as a distinct active execution. The sync worker renews its lease with a per-execution timer; the async worker uses a per-execution task. Heartbeat, completion, failure, and cleanup all remain bound to that exact execution and include the claimed job's lease_id when the server provides one. This prevents stale work from renewing or settling a replacement lease for the same job ID.
complete, fail, job heartbeats and worker heartbeats are retried on network errors, timeouts and 5xx (up to the client's retry.max_retries; they are lease-fenced, so a replay is safe - a replayed complete that already landed is answered with 409 "Job is already completed" and treated as success). An ack that still fails is reported through the worker's error event and the client's debug hook. Handlers should still be idempotent because jobs can be reclaimed after a lease expires.
- Set appropriate timeouts — Match
lease_durationto your job's expected runtime - Check abort signal — For long-running jobs, periodically check
ctx.signal.is_set(); it is set on job timeout and whenshutdown_timeoutruns out - Handle graceful shutdown — Always call
worker.stop()on SIGTERM/SIGINT, withshutdown_timeoutlonger than your longest job - Use idempotent handlers — Jobs may be retried, so handle duplicates gracefully
- Return meaningful results — Include relevant data in the return value for debugging
- Give long-lived workers a stable
worker_id— Restarts then reuse one row instead of leaving stale ones against the plan worker cap