Skip to content

README.md

Latest commit

 

History

History
211 lines (157 loc) · 7.13 KB

README.md

File metadata and controls

211 lines (157 loc) · 7.13 KB
Raw

Runway Python Worker

A worker service that executes Python code snippets on behalf of a frontend gateway. Jobs arrive over NATS JetStream; results are published back to a reply subject and the message is ACKed. Multiple workers can run in parallel against the same consumer group to scale throughput.

For an end-to-end Runway setup, start with the repository README.

Contents

Architecture

NATS JetStream (execution.requests)
  → _worker_loop()               pull batch of messages
  → _handle_message()            deserialize ExecuteSnippetRequest
  → execute_snippet()            subprocess per job (safe mode)
    or execute_snippet_unsafe()  ProcessPoolExecutor (unsafe mode)
  → _respond()                   publish ExecuteSnippetResponse to replyTo
  → _ack()                       JetStream ACK

The worker maintains a single durable pull consumer. On each loop iteration it fetches a batch (NATS_BATCH_SIZE), fans the messages out with asyncio.gather, publishes each response to the replyTo subject in the original request, then ACKs. Fetch timeouts are treated as empty batches and retried immediately; NATS errors trigger a 1-second back-off.

Execution modes (set per-job via executionType):

Mode Mechanism Isolation Timeout
safe (default) asyncio.create_subprocess_exec separate process, RLIMIT_AS memory cap asyncio.wait_for
unsafe exec() in ProcessPoolExecutor shared process space SIGALRM (Unix)

Both modes return exit code 124 on timeout and write tracebacks to stderr on unhandled exceptions.

Prerequisites

  • Python 3.12+
  • uv — install with curl -LsSf https://astral.sh/uv/install.sh | sh
  • A NATS server with JetStream enabled
  • The execution_requests JetStream stream must exist before the worker starts (the frontend gateway creates it in production)

Start a local NATS server:

docker run -p 4222:4222 nats:latest -js

Create the stream for local development using the NATS CLI:

nats stream add execution_requests \
  --subjects "execution.requests" \
  --storage memory \
  --retention limits \
  --defaults

Setup

uv sync

Install pre-commit hooks (optional, for contributors):

uv run pre-commit install

Running Locally

Start the worker:

uv run uvicorn executor.app:app --reload

The worker connects to NATS on startup and begins consuming jobs. Health check: http://localhost:8000/health.

Send a test job with the NATS CLI:

nats pub execution.requests \
  '{"jobId":"test-1","snippet":"print(\"hello\")","replyTo":"results.test-1"}'

Subscribe to see the result before publishing:

nats sub "results.test-1" &
nats pub execution.requests \
  '{"jobId":"test-1","snippet":"print(\"hello\")","replyTo":"results.test-1"}'

Docker

Build:

docker build -t runway-python-worker .

Run (pointing at a local NATS server):

docker run \
  -e NATS_URL=nats://host.docker.internal:4222 \
  -p 8000:8000 \
  runway-python-worker

The published image is available at ghcr.io/<owner>/runway-python-worker (where <owner> is the GitHub organization or user hosting this repository).

Configuration

All settings are provided via environment variables:

Variable Default Description
NATS_URL nats://localhost:4222 NATS server URL
NATS_STREAM execution_requests JetStream stream name
NATS_SUBJECT execution.requests Subject to consume from
NATS_DURABLE execution-workers Durable consumer name (shared across worker instances)
NATS_ACK_WAIT 10 Seconds before an unACKed message is redelivered
NATS_MAX_DELIVER 3 Maximum delivery attempts before a message is dropped
NATS_BATCH_SIZE 1 Messages fetched per pull request
EXEC_MEMORY_LIMIT_MB 768 Virtual memory cap per execution (MB); 0 to disable
LOG_LEVEL INFO DEBUG, INFO, WARNING, or ERROR

API

NATS: Request

Published to the execution.requests subject (or whatever NATS_SUBJECT is set to).

{
  "jobId": "unique-job-id",
  "snippet": "print('hello')",
  "timeoutMs": 30000,
  "stdin": "optional input",
  "executionType": "safe",
  "replyTo": "results.unique-job-id"
}
Field Type Required Description
jobId string yes Unique job identifier, echoed in logs
snippet string yes Python source code to execute
timeoutMs integer no Execution deadline in milliseconds (default: 30000)
stdin string no Text written to the process stdin before execution
executionType "safe" | "unsafe" no Execution mode (default: "safe")
replyTo string no NATS subject where the response is published; if absent, the result is discarded and the message is still ACKed

NATS: Response

Published to the replyTo subject.

{
  "exitCode": 0,
  "stdout": "hello\n",
  "stderr": ""
}
Field Type Description
exitCode integer Process exit code. 0 = success, 124 = timeout, 1 = unhandled exception or worker error
stdout string Captured standard output
stderr string Captured standard error; contains the traceback on exception

HTTP

Method Path Response
GET /health {"status": "healthy"} — used for liveness/readiness probes

Running Tests

Tests mock NATS and run entirely without a live server:

uv run pytest                                              # all tests
uv run pytest tests/test_main.py                           # execution logic only
uv run pytest tests/test_nats_service.py                   # NATS service only
uv run pytest -k "syntax_error or runtime_exception"       # filter by name
uv run pytest -v                                           # verbose output
File Covers
tests/test_main.py execute_snippet and execute_snippet_unsafe: success, timeout with partial output, stdin round-trip, non-blocking event loop, runtime exceptions, syntax errors, stderr capture, mixed stdout/stderr, explicit exit codes, empty snippets. Parametrized over both execution modes.
tests/test_nats_service.py NatsService lifecycle (start/stop), message routing to safe/unsafe executors, validation errors, missing reply_to, execution exceptions, NATS publish/ack error handling, multi-batch worker loop, stdin passthrough, and load_nats_settings env-var defaults.
tests/test_app.py FastAPI app lifespan and GET /health endpoint.
tests/test_models.py Pydantic model serialization/deserialization, camelCase↔snake_case conversion, ExecutionType enum, and timeout_seconds() helper.

License

MIT — see ../LICENSE.