Operate a fleet

Operations for a fleet from run a fleet.

Wind-down and drainCopy Anchor Link

Move work off a worker or one instance:

await worker.drain(deadline)     # every attached host: no new activations, evict on idle, abort all at the deadline
await worker.drain(None)         # revoke
host.wind_down(id, deadline)     # operator request for one instance; None revokes
self.lease.wind_down(deadline)   # local instance request; None yields cooperatively
worker.install_sigterm_drain()   # SIGTERM: drain (default deadline 30s), then chain to a handler installed with loop.add_signal_handler

• deadline is seconds from now. worker.drain(math.inf) drains without ever aborting.
• install_sigterm_drain() needs a running event loop: call it from a startup hook. Uvicorn installs its handler with signal.signal, which the chain cannot see. The server entry point below exits Uvicorn when the boolean worker.drained becomes true.
• While winding down, idle instances evict immediately, including instances retained by on_idle. Open streams close after placement release. Tracked execution yields at await self.eviction_checkpoint() or cancels at the deadline.
• Local instance, operator instance, local pod, and operator worker deadlines apply independently; the earliest wins. Revoking an operator request preserves local requests. Deadlines can precede the current lease expiry.
• Eviction awaits execution cleanup, tray teardown, and the final durability frontier before releasing placement. Marked durable work resumes through the Pinboard background scanner.
• Commands admitted to resident winding-down instances remain durable; tray execution resumes on the next activation. Final instance closure seals admission.
• A draining worker answers new activations 503 + Retry-After. Heartbeat acks stop extending the lease past the drain deadline. Requests for an id past its deadline answer 503 + Retry-After.
• An instance with no open stream or WebSocket for detached_timeout winds down with deadline=0 semantics: on_idle keep-alives and background tasks never extend it.
• Instances observe self.lease.eviction_deadline and on_wind_down (lifecycle).
• Operator drains and wind-downs ride heartbeat acks (drainDeadline, messages).
  • An operator drain winds every attached host down. Undrain resumes them unless the worker is draining itself.
• State: host.draining, host.drained, worker.ready, worker.superseded, host.directory.stats (activations, evictions).

Run Uvicorn with drainingCopy Anchor Link

Keep the fleet tutorial's app.py, including its store setup, app lifespan, and worker configuration. Save this beside it as serve.py:

import asyncio
import os
from contextlib import asynccontextmanager, suppress

import uvicorn

from app import app, worker


async def main():
    server = uvicorn.Server(
        uvicorn.Config(app, host="0.0.0.0", port=int(os.environ.get("PORT", "8000")))
    )
    original_lifespan = app.router.lifespan_context

    async def exit_after_drain():
        while not worker.drained:
            await asyncio.sleep(0.1)
        server.should_exit = True

    @asynccontextmanager
    async def lifespan(app):
        async with original_lifespan(app) as state:
            worker.install_sigterm_drain(deadline=30)
            shutdown = asyncio.create_task(exit_after_drain())
            try:
                yield state
            finally:
                shutdown.cancel()
                with suppress(asyncio.CancelledError):
                    await shutdown

    app.router.lifespan_context = lifespan
    await server.serve()


if __name__ == "__main__":
    asyncio.run(main())

Install Uvicorn with uv add uvicorn. Keep the fleet tutorial's worker environment variables, and replace its Uvicorn command with:

PORT=8000 uv run python serve.py

Use PORT=8001 for the second worker and match each ADVERTISE_URL to that port. Inside this repo, run uv run --directory python python "$PWD/serve.py". Run one server process per advertise URL.

Verify the shutdownCopy Anchor Link

1. Start a streaming reply through pinboard, then send kill -TERM <PID> to its owning server process. Use the PID from Uvicorn's startup log.
2. The worker enters draining and refuses new activations. Existing work can finish until the drain deadline; remaining instances close at that deadline, subject to the current lease.
3. After every attached host is empty, worker.drained becomes true. The watcher requests Uvicorn shutdown, the original app lifespan finishes, and the process exits.
4. Keep the other worker running. Reattach through pinboard and confirm the stored thread is available there.

SIGTERM exercises the graceful drain path. The owner-crash test uses SIGKILL and exercises recovery from stored work instead. An operator drain also makes this entry point exit once the worker is empty.

Operator drainCopy Anchor Link

Cordon and release a worker by client id:

curl http://pinboard.internal:8080/registry/workers \
  -H "Authorization: Bearer $PINBOARD_OPERATOR_TOKEN"     # {"workers":[{"clientId","advertiseUrl","sessionId","hostname","namespaces","envs",...}]}
curl -X POST http://pinboard.internal:8080/registry/drain \
  -H "Authorization: Bearer $PINBOARD_OPERATOR_TOKEN" \
  -d '{"clientId": "<from /registry/workers>", "deadline": null}'      # cordon
curl -X POST http://pinboard.internal:8080/registry/undrain \
  -H "Authorization: Bearer $PINBOARD_OPERATOR_TOKEN" \
  -d '{"clientId": "<from /registry/workers>"}'

• Match a worker to its client id through advertiseUrl in GET /registry/workers; GET /registry/stats carries the same clientId.
• With the fleet page's single-key passport, $PINBOARD_OPERATOR_TOKEN is $PINBOARD_TOKEN; a deployment splits the operator key out (passport).
• deadline is relative seconds; a value under one lease TTL becomes one lease TTL. null cordons: no new placements, and the attached hosts wind down, idle instances at once and work in flight at its own deadline.
• A cordon holds until POST /registry/undrain with the same client id, surviving re-registration. A restarted process mints a new client id, and the old cordon key stays in Redis until undrained.
  • Timed drains expire on their own 7 days past the deadline.
• Both routes need pinboard:operator whose claims.pinboard covers the worker's namespaces, hostname, and env; otherwise 403 {"error":"forbidden"}. A client id that never registered against this keyspace answers the same 403. A registered worker that is down still takes the drain: the deadline lives in its client config and applies when it re-registers.

Rolling deployCopy Anchor Link

1. Start the replacement workers. They register and take new placements.
2. Cordon each old worker: POST /registry/drain with {"clientId": ..., "deadline": null}.
3. SIGTERM any old pods still running. The server entry point drains and exits when worker.drained becomes true; a cordoned worker may already have emptied and exited.
4. Old workers empty and exit. Open attaches close with fin: reconnect, code: evicted; clients reconnect and the ids re-activate on the new deployment (evicted).
5. POST /registry/undrain every cordoned client id, clearing their keys.

• Size terminationGracePeriodSeconds above the drain deadline (default 30s) plus one lease TTL (15s), plus uvicorn's own shutdown.
• POST /registry/wind_down moves one id, or every id in the (env, namespace) when id is omitted. Shapes: pinboard registry.
• A reconfigured pinboard fleet (PINBOARD_SHARDS, PINBOARD_HOSTNAME_AWARE) rolls with a higher PINBOARD_EPOCH.
  • The new license displaces the old fleet (fleet license). A displaced node fails readyz; the Node entry exits 1 through onDisplaced (environment).

Hostname-aware modeCopy Anchor Link

One fleet serving many hostnames (preview deployments, per-tenant domains), enabled by PINBOARD_HOSTNAME_AWARE. Workers then carry a hostname and an env:

from pinned import Host, PinnedWorker

worker = PinnedWorker(                        # reads PINBOARD_URL, PINBOARD_TOKEN, ADVERTISE_URL
    hosts={"/threads": Host(Counter, env="staging")},
    hostname="pr-42.preview.example.com",     # required in hostname-aware mode
    env="pr-42",                              # defaults to the hostname
    takeover="wind_down",                     # "wind_down" (default) or "deny"
    epoch=1,                                  # displaces lower-epoch bindings
)

• Types and defaults: PinnedWorker. Host(Api, env=...) overrides the env for one namespace.
• A hostname with no live workers answers 503 no live workers for hostname + Retry-After. A missing Host answers 400 {"error":"missing Host header"}.
• The first registration binds (hostname, namespace) -> env and env -> takeover. A disagreeing registration at the same or a lower epoch is 409:
  • hostname X/ns is bound to env Y at epoch N, not Z — register with a higher epoch to displace
  • env X agreed takeover Y at epoch N, not Z — register with a higher epoch to displace
• Bindings and their epochs are garbage-collected with their last live worker.
• Cluster calls pin Host to the caller's worker's hostname.
• Worker passport keys need claims.pinboard.hostnames and claims.pinboard.envs (Passport):

{
  "providers": [{ "type": "staticToken", "id": "pinboard", "keys": [
    { "name": "previews", "token": [{ "env": "PINBOARD_TOKEN" }, { "env": "PINBOARD_TOKEN_NEXT" }],
      "scopes": ["pinboard:worker"],
      "claims": { "pinboard": { "namespaces": ["/threads"], "allowedBackendHosts": ["10.0.0.0/8"],
        "hostnames": [".preview.example.com"], "envs": ["*"] } } },
    { "name": "ops", "token": { "env": "PINBOARD_OPERATOR_TOKEN" }, "scopes": ["pinboard:operator"],
      "claims": { "pinboard": { "namespaces": ["*"], "hostnames": ["*"], "envs": ["*"] } } }
  ]}],
  "namespaces": { "/threads": { "allow": "authenticated" } }
}

• Every env var a token array names must be set; a missing one fails startup with PINBOARD_PASSPORT is invalid. Rotate by swapping the two values.
• Single mode rejects hostname, env, envs, takeover, and a nonzero epoch.

TakeoverCopy Anchor Link

A request for an id placed under another hostname in the same env applies the env's takeover:

• Under wind_down, sustained traffic from two hostnames swings the id back and forth, one lease per swing.

Epoch displacementCopy Anchor Link

• A registration whose epoch exceeds the bound one rewrites a disagreeing binding.
  • The displaced (hostname, namespace) refuses new placements for one lease (hostname binding displaced, retry). Ids already placed under the new env keep serving.
• Displaced holders get 409 {"error": ..., "superseded": true} on their next heartbeat: PinnedWorker aborts its instances, stops registering, and reports worker.superseded.
• Epochs are monotonic per binding. Rollback is a fresh, higher epoch. Any key whose claims admit the hostname and env may displace.

ObservabilityCopy Anchor Link

• pinned: lifecycle events (activation, eviction/abort, drain, lease health, registration) log on the pinned.* logger hierarchy with an instance_id field. An attach denial is unlogged; the response body carries AuthRejected's message or payload. Log it in authorize.
  • With prometheus_client installed, importing pinned registers the pinned_* collectors on the default registry. The app serves them:

from prometheus_client import make_asgi_app

app.mount("/metrics", make_asgi_app())   # open to anyone who reaches the pod unless the app gates it

• pinboard: pino JSON on stdout at LOG_LEVEL. Hostname and env appear only as log fields.
  • GET /metrics (pinboard:operator with unrestricted claims, Node entry) serves the pinboard_* collectors. Scrape every node.
  • Metric names, aggregation, and the logger and metrics hooks: pinboard reference.

Scrape both sides:

scrape_configs:
  - job_name: pinboard
    authorization: { credentials: "<pinboard:operator token, unrestricted claims>" }
    static_configs: [{ targets: ["pinboard-1:8080", "pinboard-2:8080", "pinboard-3:8080"] }]
  - job_name: pinned-workers
    static_configs: [{ targets: ["10.0.3.7:8000", "10.0.3.8:8000"] }]

Alert on a fleet losing its workers:

groups:
  - name: fleet
    rules:
      - alert: WorkerLeaseUnhealthy      # heartbeat acks stopped; this worker refuses new activations
        expr: pinned_lease_healthy == 0
        for: 1m
      - alert: NamespaceWithoutWorkers
        expr: increase(pinboard_placement_resolutions_total{outcome="no-backends"}[5m]) > 0
      - alert: WorkerUnreachable
        expr: increase(pinboard_forward_failures_total{outcome="unreachable"}[5m]) > 0

• Pinboard liveness and readiness: GET /healthz, GET /readyz.

Worker readiness and recoveryCopy Anchor Link

Add a readiness route to the worker app:

from fastapi import Response

@app.get("/readyz")
async def readyz() -> Response:
    return Response(status_code=200 if worker.ready and not worker.superseded else 503)

This checks registration and lease expiry. The worker's request handler separately gates new activations on lease health and draining.

• worker.ready describes current serving eligibility; worker.lease.healthy reports the earlier heartbeat health boundary. Observe both with the lease metrics.
• worker.draining and worker.drained report the shutdown path separately. The Uvicorn entry point exits after a drain empties the worker.
• Retire a superseded process through your deployment or process supervisor. Supersession alone leaves the Python server process running; it is a permanent registration failure for that worker.
• In Kubernetes, readiness failure removes a pod from matching Service endpoints. Restart decisions come from liveness/startup failures and the container restart policy. Use readiness for recoverable lease failures and an explicit termination policy for superseded workers. Kubernetes probe behavior.

Inspect the workersCopy Anchor Link

• GET /registry/stats collects each worker's /_pinned/stats (that route answers pinboard only, on the shared secret or an attestation) and reports draining, leaseFlaps, reregistrations, per-namespace activations/evictions, per-instance connections/tasks/idleMs/activatedAt, and process uptimeMs/rssBytes/cpuPercent (psutil).

Admin consoleCopy Anchor Link

apps/admin is a Next.js app: pnpm dev on 127.0.0.1:3003. / redirects to /inspector.

• The browser talks to same-origin proxies: /api/pinboard/* attaches pinboard credentials, /api/wire/* carries Statewire traffic. Both reject upstream redirects.
• ADMIN_PROXY_ALLOWED_ORIGINS (comma-separated exact origins) restricts targets. Left undefined, development (NODE_ENV=development) accepts any HTTP target and every other environment answers 403 {"error":"proxy target is not allowed"} for all of them.

Connections come from the environment:

PINBOARD_CONNECTIONS='[
  { "name": "prod", "url": "https://pinboard.example.com", "token": "<pinboard:operator token>" },
  { "name": "sso", "url": "https://pinboard.internal", "forward": true }
]'

• An entry carries exactly one of token (sent as Authorization: Bearer) and forward (true forwards the inbound authorization header; a header name forwards that header).
  • GET /api/connections exposes { name, url }, or 500 { error } on a malformed value.
  • The browser sends x-pinboard-connection: <name>, or x-pinboard-url / x-pinboard-token for a custom connection.
• Fleet (/fleet, /fleet/hosts, /fleet/instances) polls GET /registry/overview, /registry/stats, and /registry/health every 3s.

  • Shows host, instance, and connection counts, hosts draining, degraded, or winding down, and control-plane resolution counters.

  • Hosts offer Cordon (null deadline), Force drain in 30s / 5m / 30m, Undrain.

  • Instances offer Wind down (deadline 0), Cancel wind-down (null), and Inspect.

    • Hostname-aware mode requires env in the POST /registry/wind_down body: 400 {"error":"env is required in hostname-aware mode"} without it. Single mode rejects env: 400 {"error":"env requires hostname-aware mode"}.

    Copy Text

    { "namespace": "/threads", "id": "alice", "deadline": 0, "env": "pr-42" }

  • The hostname dropdown scopes the views (persisted in localStorage, reset on connect).

  • ?inspect= opens a host or instance drawer; Inspect deep-links pinboard-URL/namespace/id into the inspector.

• Inspector (/inspector?url=) attaches to any Statewire instance with POST /stream and writes frames via POST /frames; it works standalone.