Skip to content

CLAUDE.md

Latest commit

 

History

History
133 lines (78 loc) · 21 KB

CLAUDE.md

File metadata and controls

133 lines (78 loc) · 21 KB
Raw

CLAUDE.md

This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.

Project Overview

vgi-rpc is a transport-agnostic RPC framework built on Apache Arrow IPC serialization. RPC interfaces are defined as Python Protocol classes; the framework derives Arrow schemas from type annotations and provides typed client proxies with automatic serialization/deserialization.

Commands

# Run all tests (includes mypy type checking and ruff linting via pytest plugins)
pytest

# Run a single test
pytest tests/test_rpc.py::test_name

# Lint and format
ruff check vgi_rpc/ tests/
ruff format vgi_rpc/ tests/

# Type checking
mypy vgi_rpc/

# Coverage (80% minimum, branch coverage enabled)
pytest --cov=vgi_rpc

Uses uv as the package manager. Install dev dependencies with uv sync --all-extras.

Tests should complete in 50 seconds or less ALWAYS!

Discourage the use of Any types, check mypy strict type coverage and always try to improve it.

Before pushing changes make sure, mypy, ruff and tests pass.

Pay attention to mypy strict type checking make sure strict typing is preserved.

Verify "ty" type checking too.

The full process before committing code is

  1. Run uv run ruff format . on all files
  2. Run uv run ruff check . and resolve all errors
  3. Run uvx --python 3.13 --from pydoclint==0.8.7 pydoclint vgi_rpc/ and resolve all errors (docstring args/attributes must match the code; config in [tool.pydoclint], no baseline — the tree is fully clean). pydoclint runs isolated via uvx because its docstring-parser-fork dep collides with the canonical docstring-parser in the project venv — so it is intentionally not a project dependency.
  4. Run uv run mypy vgi_rpc/ and resolve all errors
  5. Run uv run ty check vgi_rpc/ and resolve all errors
  6. Run uv run pytest for all tests

Always reformat before pushing. CI runs lint before tests — unformatted code fails the pipeline immediately and wastes time.

Architecture

Core modules (vgi_rpc/)

  • rpc/ — The RPC framework package. Defines the wire protocol, method types (UNARY, STREAM), and the core classes: RpcServer, RpcConnection, RpcTransport, PipeTransport, ShmPipeTransport, SubprocessTransport, UnixTransport, TcpTransport, StreamSession, AnnotatedBatch, OutputCollector. Also defines AuthContext (frozen dataclass for authentication state), CallContext (request-scoped context injected into methods via ctx parameter), _TransportContext (contextvar bridge for HTTP auth), RpcError, and VersionError. Introspects Protocol classes via rpc_methods() to extract RpcMethodInfo (schemas, method type). Client gets a typed proxy from RpcConnection; server dispatches via RpcServer.serve(). The convenience function run_server() parses sys.argv for --http, --host, --port, --unix, and --tcp flags — without any it serves over stdin/stdout pipes (for SubprocessTransport); with --http it lazily imports vgi_rpc.http.serve_http and serves over HTTP; --unix PATH and --tcp [HOST:]PORT serve the raw Arrow-IPC framing over a Unix or TCP socket (serve_unix / serve_tcp), each emitting a UNIX:<path> / TCP:<host>:<port> discovery line on stdout. The TCP transport defaults the host to loopback (127.0.0.1) and carries no auth/TLS — it is for trusted networks only; use HTTP otherwise. The _debug.py submodule provides wire protocol debug logging infrastructure: 6 logger instances under the vgi_rpc.wire.* hierarchy (request, response, batch, stream, transport, http) and 4 formatting helpers (fmt_schema, fmt_metadata, fmt_batch, fmt_kwargs). All log points use isEnabledFor guards for zero overhead when disabled.

  • utils.py — Arrow serialization layer. ArrowSerializableDataclass mixin auto-generates ARROW_SCHEMA from dataclass field annotations and provides serialize()/deserialize_from_batch(). Handles type inference from Python types to Arrow types (including generics, Enum, Optional, nested dataclasses). Also provides low-level IPC stream read/write helpers, IpcValidation enum, and ValidatedReader.

  • log.py — Structured log messages (Message with Level enum). Messages are serialized out-of-band as zero-row batches with metadata keys vgi_rpc.log_level, vgi_rpc.log_message, vgi_rpc.log_extra. Server methods access logging via the CallContext (see below).

  • logging_utils.py — VgiJsonFormatter, a logging.Formatter subclass that serializes log records as single-line JSON. Not auto-imported; must be imported explicitly from vgi_rpc.logging_utils.

  • metadata.py — Shared helpers for pa.KeyValueMetadata. Centralises well-known metadata key constants (vgi_rpc.method, vgi_rpc.stream_state#b64, vgi_rpc.log_level, vgi_rpc.log_message, vgi_rpc.log_extra, vgi_rpc.server_id, vgi_rpc.request_version, vgi_rpc.location, vgi_rpc.shm_offset, etc.) and provides encoding, merging, and key-stripping utilities used by rpc/, http/, log.py, external.py, shm.py, and introspect.py.

  • introspect.py — Introspection support. Provides the built-in __describe__ RPC method, MethodDescription, ServiceDescription, build_describe_batch, parse_describe_batch, compute_protocol_hash, and introspect(). Enabled on RpcServer via enable_describe=True. The wire format is DESCRIBE_VERSION = "4" — slim 8-column schema (name, method_type, has_return, params_schema_ipc, result_schema_ipc, has_header, header_schema_ipc, is_exchange). Python-flavoured fields (doc, param_types_json, param_defaults_json, param_docs_json) were dropped in v4 so the wire stays language-neutral; the Protocol source class is the source of truth for human-readable type names, defaults, and docstrings. The response batch's custom metadata also carries vgi_rpc.protocol_hash — a SHA-256 hex digest over the canonical describe payload that uniquely identifies the protocol contract within a process and is stable across runs/builds for the same Protocol.

  • access_log.schema.json + access_log_conformance.py — Cross-language access-log spec. Access logging is an HTTP-transport concern only. Pipe, subprocess, shared-memory pipe, and Unix-socket transports do not emit access logs — those transports run trusted, co-located worker processes where per-call audit logging adds no value. The --access-log flag and _configure_access_log helper are only meaningful when serving over HTTP (e.g. vgi-rpc-test --http --access-log ... or serve_http(..., access_log=...)). Every conformant HTTP server emits one JSON record per RPC call on the vgi_rpc.access logger; the schema is enforced by vgi-rpc-test --access-log <path>. Always-required fields include protocol_hash (the hash from __describe__) so consumers reading archived JSONL can decide whether a cached schema decoder still applies. Records carry truncated: true (or truncated: "record_too_large") when the formatter shed fields to stay under --access-log-max-record-bytes (default 1 MiB). Rotation is via --access-log-max-bytes (size) or --access-log-when (time); paths support {pid} and {server_id} placeholders so multiple HTTP workers in one container don't collide. Reference shipper configs (Vector + Fluent Bit, S3/GCS/Azure) live under docs/log-shipping/. See docs/access-log-spec.md for the full contract and docs/porting-guide.md for the cross-language conformance status of Go/TypeScript/Java/Rust ports.

  • shm.py — Shared memory transport support. Provides ShmAllocator, ShmSegment, and pointer batch helpers for zero-copy Arrow IPC batch transfer between co-located processes. Used by ShmPipeTransport.

  • pool.py — Subprocess process pool with shared memory support. WorkerPool keeps idle worker subprocesses alive between connect() calls, avoiding repeated spawn/teardown overhead. Workers are keyed by command tuple, cached up to max_idle globally (LIFO for cache warmth), and evicted by a daemon reaper thread after idle_timeout. When shm_size is set on the pool, each connect() borrow gets its own isolated ShmSegment that is automatically created and destroyed per borrow. Tracks _stream_opened / _last_stream_session to detect abandoned streams and discard workers with stale transport state. Provides PoolMetrics for observability. Logger: vgi_rpc.pool.

  • external.py — ExternalLocation batch support for large data. When batches exceed a configurable size threshold, they are uploaded to pluggable ExternalStorage (e.g. S3) and replaced with zero-row pointer batches containing a vgi_rpc.location URL metadata key. Readers resolve pointers transparently via external_fetch.fetch_url() (aiohttp-based parallel fetching); writers externalize batches above the threshold. Provides ExternalLocationConfig, ExternalStorage protocol, UploadUrl, UploadUrlProvider, Compression enum, https_only_validator, and production/resolution functions. Supports optional zstd compression.

  • external_fetch.py — Parallel range-request URL fetching. Issues a HEAD probe to learn Content-Length and Accept-Ranges, then either fetches in parallel chunks with speculative hedging for stragglers, or falls back to a single GET. Maintains a persistent aiohttp.ClientSession per FetchConfig on a daemon thread. Handles zstd decompression and stale-connection recovery. Provides FetchConfig and fetch_url().

  • cli.py (optional — pip install vgi-rpc[cli]) — Command-line interface. A typer-based CLI registered as the vgi-rpc entry point. Provides describe, call, and loggers commands for introspecting and invoking methods on any vgi-rpc service. Supports output formats auto, json, table, and arrow (raw Arrow IPC via --format arrow), with --output/-o for file output. Stream headers are surfaced in all formats.

  • s3.py (optional — pip install vgi-rpc[s3]) — S3 storage backend implementing ExternalStorage. Uses boto3 to upload IPC data and generate pre-signed URLs. Supports custom endpoints for MinIO/LocalStack.

  • gcs.py (optional — pip install vgi-rpc[gcs]) — Google Cloud Storage backend implementing ExternalStorage. Uses google-cloud-storage to upload IPC data and generate V4 signed URLs. Relies on Application Default Credentials.

  • http/ (optional — pip install vgi-rpc[http]) — HTTP transport package using Falcon (server) and httpx (client). Exposes make_wsgi_app() to serve an RpcServer as a Falcon WSGI app, serve_http() as a convenience wrapper that combines make_wsgi_app + automatic free-port selection + waitress.serve (prints PORT:<port> to stdout for machine-readable discovery), and http_connect() for the client side. Streaming is stateless: each exchange carries serialized StreamState in a signed token in Arrow custom metadata. Supports pluggable authentication via an authenticate callback and _AuthMiddleware. Includes _testing.py with make_sync_client() for in-process testing without a real HTTP server. Sticky sessions (HTTP-only, opt-in via make_wsgi_app(enable_sticky=True)) live in _sticky.py — registry, reaper, middleware, AEAD session token sealing, and the DELETE /vgi/__session__ synthetic endpoint. The runtime API (CallContext.open_session / close_session / session / session_id) lives in rpc/_common.py so it's available to type-check on every CallContext regardless of transport, but raises RuntimeError when invoked on non-HTTP transports. Client side: HttpConnection.with_session_token() context manager yields a _SessionView that auto-injects VGI-Session-Accept: true + VGI-Session: <token> on every request inside the block and captures responses; view.detach() hands the token off for stashing across process lifecycles.

    HTTP response caps. Two independent operator knobs gate response size:

    • max_response_bytes caps the HTTP body (what literally lands on the wire). Default None = unbounded. For producer streams this is soft — continuation tokens cover overshoot; for unary and stream-exchange it is hard and surfaces as RpcError (200 + X-VGI-RPC-Error: true + EXCEPTION batch).
    • max_externalized_response_bytes caps total bytes uploaded to external storage during one HTTP response. Always hard — externalised uploads have no escape valve. Strict-fail surfaces the same way.

    Externalised payloads are NOT charged against max_response_bytes (they leave only tiny pointer batches on the wire). The two knobs answer different operator questions: HTTP body size (proxy/gateway limit) vs per-call data volume.

    Both are surfaced via response headers (VGI-Max-Response-Bytes, VGI-Max-Externalized-Response-Bytes, VGI-Externalization-Enabled) so http_capabilities() and conformance tests can probe them. Workers can read out.remaining_response_bytes / out.remaining_externalized_response_bytes / out.externalization_enabled on OutputCollector to size emits within budget.

    The deprecated alias max_stream_response_bytes (constructor kwarg, --max-stream-response-bytes CLI flag, VGI_RPC_MAX_STREAM_RESPONSE_BYTES env) is retained for one release cycle; emits a DeprecationWarning when set. The cap is no longer stream-only — it now governs all HTTP method responses.

Wire protocol

Multiple IPC streams are written sequentially on the same pipe. Every request batch carries vgi_rpc.request_version in custom metadata; the server validates this before dispatch and rejects mismatches with VersionError. Each method call writes one request stream and reads one response stream:

  • Unary: Client sends params batch → Server replies with log batches + result/error batch
  • Stream: Initial params exchange, then lockstep: client sends input batch (tick for producer, real data for exchange), server replies with log batches + output batch, repeating until EOS

For HTTP transport, the wire protocol maps to separate endpoints: POST /vgi/{method} (unary), POST /vgi/{method}/init (stream init), POST /vgi/{method}/exchange (stream exchange).

Key patterns

Defining an RPC service: Write a Protocol class where return types determine method type — plain types for unary, Stream[S] for streaming (both producer and exchange patterns).

Stream state: Streaming methods return a Stream[S] where S is a StreamState subclass. The state's process(input, out, ctx) method is called once per iteration. Producer streams (default input_schema=_EMPTY_SCHEMA) ignore the input and call out.finish() to end. Exchange streams set input_schema to a real schema and process client data.

CallContext injection: Server method implementations can accept an optional ctx: CallContext parameter. CallContext provides auth (AuthContext), client_log() (client-directed logging), emit_client_log (raw ClientLog callback), transport_metadata (e.g. remote_addr from HTTP), kind (the active TransportKind), and a logger property returning a LoggerAdapter with request context pre-bound. The parameter is injected by the framework — it does not appear in the Protocol definition.

Transport awareness: Workers can read RpcServer.transport_kind (a TransportKind enum: PIPE, HTTP, UNIX, TCP) and RpcServer.transport_capabilities (currently {"shm"} when bound to a ShmPipeTransport). For one-shot startup work an implementation may define an on_serve_start(self, kind: TransportKind) -> None method (the ServeStartHook Protocol); the framework calls it once per process before the first dispatch. For pipe/unix the hook fires inside RpcServer.serve(transport); for HTTP it fires lazily on the first request handled in the current process so pre-fork servers (gunicorn, uwsgi) correctly run it in each child. Hook exceptions propagate (and are logged via logging.getLogger("vgi_rpc.rpc").exception first); rebinding to a different kind re-fires the hook rather than raising. Per-call code can also branch on ctx.kind.

Authentication: AuthContext (frozen dataclass) carries domain, authenticated, principal, and claims. For HTTP transport, make_wsgi_app(authenticate=...) installs _AuthMiddleware that calls the callback on each request and populates CallContext.auth. Pipe transport gets anonymous auth by default. Methods can call ctx.auth.require_authenticated() to gate access.

Server identity: Each RpcServer gets a server_id (auto-generated 12-char hex or caller-supplied). This ID is attached to all log and error batches as vgi_rpc.server_id metadata for distributed tracing. RpcServer also accepts enable_describe=True to register the synthetic __describe__ introspection method.

Protocol identity: RpcServer computes protocol_hash (SHA-256) over the canonical __describe__ payload at construction. Three answer-different questions:

  • server_version (constructor arg, free-form): which build produced this record. Used for rollout/rollback dashboards.
  • protocol_version (read from vars(protocol).get("protocol_version") — ClassVar[str] on the Protocol class, canonical semver MAJOR.MINOR.PATCH): which release of the protocol surface contract this is. Enforced: when the Protocol declares it, the client sends it on every request as vgi_rpc.protocol_version custom_metadata and the server raises ProtocolVersionError at the dispatch boundary on an exact major+minor mismatch (patch ignored). __describe__ is exempt so a mismatched client can introspect to discover the server's version. Carries opt-in semantics — Protocols that don't declare it disable the check entirely.
  • protocol_hash (auto-computed): byte-stable identity of the Protocol from the Python codegen's POV. Surfaced in __describe__ response metadata and on every access-log record. Not cross-language byte-stable (Arrow IPC schema serialization isn't guaranteed identical across pyarrow / arrow-cpp / arrow-rust). Treat as a Python-internal drift detector; the cross-language contract is protocol_version.

Error propagation: Server exceptions become zero-row batches with error metadata; clients receive RpcError with error_type, error_message, and remote_traceback. The transport stays clean for subsequent requests.

Sticky sessions (HTTP, opt-in): With make_wsgi_app(enable_sticky=True), a method body may call ctx.open_session(state) to register a Python object (DB cursor, model handle, file handle) in a per-worker registry; subsequent calls from the same client (inside conn.with_session_token():) carry a VGI-Session header that resolves back to the object as ctx.session. Eviction is TTL-driven (default 300s, override per-call via ttl=) or explicit (ctx.close_session()); state.close() is invoked on eviction if defined. The framework serializes concurrent calls on the same session via a per-session RLock; different sessions run in parallel. Misroute / expiry / cross-principal token surfaces as SessionLostError (typed, error_kind="session_lost"); drain-time opens surface as ServerDrainingError. Client uses conn.with_session_token() as sess: (auto-sends VGI-Session-Accept: true for the server's leak-prevention guard) and sess.detach() to stash a token for later resumption without firing the exit-time best-effort DELETE /vgi/__session__. Sticky machinery is not installed on pipe / subprocess / unix transports; ctx.open_session raises RuntimeError there. Echo headers (sticky_echo_headers= kwarg on make_wsgi_app): tell the client to replay arbitrary headers on every subsequent request in the session — emitted as VGI-Echo-<name>: <value> on the session-opening response, captured + replayed by the client view, exposed via sess.current_echo_headers(). Used for client-driven routing on platforms like Fly.io; vgi_rpc.http.fly.fly_sticky_echo_headers() + vgi_rpc.http.fly.auto_server_id() are the ~25-line Fly quickstart helpers (return None off Fly so the same code works everywhere). Graceful drain: drain_handle(app) returns DrainHandle(drain, shutdown, is_draining) for operator-facing shutdown wiring; serve_http(enable_sticky=True) auto-installs SIGTERM/SIGINT handlers that drain → wait drain_grace_seconds → invoke state.close() on every live session → exit. Pre-fork servers (gunicorn) use drain_handle(app) inside worker_exit hooks. Access log carries session_id (24-char hex) + session_action (none/open/resume/close) on sticky-touching records; both absent on non-sticky servers. Full spec: docs/sticky-sessions-spec.md. Cross-language conformance group: TestSticky in vgi_rpc/conformance/_pytest_suite.py, capability-gated on VGI-Sticky-Enabled; TestSticky::test_echo_header_round_trip further capability-gated on VGI-Sticky-Echo-Headers; TestSticky::test_drain_rejects_new_opens further capability-gated on the conformance server exposing POST /__test_drain__ admin endpoint.

Code Style

  • Line length 120, double quotes, target Python 3.13+
  • Strict mypy (python_version = "3.13", strict = true)
  • Ruff rules: E, F, I, UP, B, SIM, D, RUF, PERF (includes docstring enforcement)
  • Google-style docstrings with Args/Returns/Raises sections