Console & REST API¶

Web Console¶

The console is a browser UI over the REST API described below: node health and partition assignments, the application's topology, live window contents, the operational event feed, and dead-letter inspection. It is observation, with one exception: in Raft mode it can also change cluster membership (see Node actions). Nothing in it edits your data or your app's configuration.

Open it¶

Every node serves the console itself, on the same port as the API:

http://<node-host>:8400/console

There is nothing to install, build or deploy — the console ships inside the Turbine package, so a node you can already reach on its API port already has it. The default port is 8400, configurable via api_port; the node's bare root (http://<node-host>:8400/) redirects here.

It is a single-page app that runs entirely in your browser and calls the API directly. There is no console backend and no proxy in the path, which is why it needs no configuration of its own: the node that served the page is the node it queries.

Hosting the console yourself

You can serve the bundle from your own web server instead — behind an existing authenticating proxy, for example. This needs a source build: the installed package embeds the console inside the compiled extension, so there are no files on disk to point a web server at. From a source checkout, just console-build writes the bundle to console/dist/.

Serve it under a /console/ path on that origin: its asset links are absolute, and any unmatched path below it must return index.html, or reloading the page on a view like /console/topology gives you your web server's 404 instead of the page. In nginx that is try_files $uri /console/index.html;. Hosted this way the console has no node of its own to query, so add your endpoints in Settings on first use.

Point it at more nodes¶

Out of the box the console queries the node it was served from, which is all a single-node deployment needs. To watch several nodes, open Settings and add the base URL of each one's API, for example http://10.0.0.1:8400. The list is stored in your browser, per browser and per machine; it is not shared between users and there is nothing to deploy when it changes.

Entries can be individually enabled or disabled. Reads initially try them top to bottom, skipping nodes with network failures or server errors. After a successful fallback, subsequent polls start at that node, avoiding another wait for the unavailable primary. If it fails, the other nodes are tried in configured order. There is no periodic return to the primary while the fallback works; apply Settings or reload the page to retry the configured order immediately. Each attempt allows up to 30 seconds for the response, including its body. A node answering "no" to a request — an unknown window, a topic that isn't a dead-letter topic — ends the request there rather than asking every other node the same question. Mutation actions always use the first enabled node in configured order and are sent only once, even after a timeout or lost response.

Because the browser talks to each node directly, every node you list must be reachable from the browser, not merely from the node serving the page. On Kubernetes the pods are not: see Opening the console.

You do not need a cluster. Against a single standalone node the cluster views simply show one node. What you gain by running Raft is that any one node answers for all of them, so one endpoint is enough to see the whole cluster.

What the views show¶

View	Shows
Nodes	Every node with its workers: throughput, lag, state operations, restarts, recent dead-letters. Pivot the same facts by node (where things run) or by subscription (what your app does). Selecting a worker opens a detail panel, and you can keep several open as tabs. In Raft mode, this is also where the node actions live.
Topology	Your application as a graph — topics, subscriptions, repartition hops — drawn from the `@subscribe` declarations and annotated with live rates. Where you see key skew across partitions, or a repartition hop falling behind.
Windows	Open persistent windows (tumbling, sliding, session) and their contents. Windows are listed by their label when they have one, and filtered by name or label — an app that builds one window per rule lists hundreds. Pick a group and you get its curve: the value of each pane, with the current window highlighted and its aggregate as a guide line. The fastest way to answer "why did this fire and its neighbour not?".
Timers	Current persistent timers by origin (window, join expiry, user alarm). Interval joins show retained left/right rows, pending expiry buckets, their close frontier, late rows and input skew per worker. Worker counts link here for join expiries and user timers; window counts link to the Windows views.
Events	The operational feed: worker starts, stops, failures, restarts. This is what makes a crash-looping worker visible; a throughput gauge stuck at zero cannot tell you the difference between a crash-loop and an idle topic.
DLQ	Per-subscription dead-letter configuration and counters, plus a non-destructive peek at the actual rejected records and why they were rejected.

Views poll every few seconds, and polling pauses while the browser tab is in the background, so leaving the console open on a second screen does not keep loading your nodes.

Node actions¶

In Raft mode only, the Nodes view offers three membership operations, each behind a confirmation:

Exclude from cluster — take a node out of the voting set so you can stop it for maintenance without the cluster waiting on it. Its partitions are reassigned to the remaining nodes. Refused when removing that node would break quorum, so you cannot lose availability by clicking it.
Include into cluster — put an excluded node back in.
Transfer leadership — hand the leader role to another node, typically before taking the current leader down.

These are real cluster operations with the same effect as the corresponding REST calls below; the console is a convenience over them, not a separate mechanism. A standalone node shows no such buttons.

Securing the API¶

Out of the box the API has no authentication: anyone who can reach a node's API port can read your state, your window contents and your dead-lettered records — which routinely contain production payloads — and, in Raft mode, can change cluster membership. Turbine says so at startup (API auth: ⚠ NONE in the banner, and a warning in the log). That default keeps a test cluster free of any setup; do not keep it past a private network.

Set a token with TURBINE_API_TOKEN (or Turbine(api_token=...)), the same value on every node of a cluster:

export TURBINE_API_TOKEN="$(openssl rand -hex 32)"   # once; share it with every node

With a token set:

Every route requires Authorization: Bearer <token> — reads, writes and /metrics alike — and answers 401 without it. /metrics also accepts a separately configured scrape token, described below.
/health and /ready stay open, so orchestrator probes need no configuration, and so does the console page itself (/console), which asks for the token the first time a request is refused and keeps it in your browser. You can also enter it in Settings.
Nodes send the token to each other on every call (Raft replication, requests forwarded to the leader, cluster-wide views). A node configured with a different token is reported as such, not as unreachable: cluster-wide answers list it under unauthorized, the Nodes view marks it unauthorized, and joining the cluster fails with an error naming the token.

curl -H "Authorization: Bearer $TURBINE_API_TOKEN" http://localhost:8400/assignments

What a token does not do:

It is not encryption. The API speaks plain HTTP, so the token and the data travel in clear: keep the port on a network you trust, or put a TLS-terminating proxy in front of it.
The API token grants all API rights. For Prometheus, configure a separate TURBINE_METRICS_TOKEN (Turbine(metrics_token=...)), accepted only on /metrics. It requires a distinct admin token; it cannot read application state or change cluster membership or readiness. See Scraping the metrics. There are no per-user credentials or general read-only API credentials.
Changing the API token means restarting every node. Nodes with different API tokens cannot talk to each other, so a rolling change leaves the cluster split until the last node has restarted.

Two more settings narrow the exposure:

TURBINE_API_HOST (api_host=) — the address the API listens on, 0.0.0.0 by default. 127.0.0.1 keeps it local to the machine; do not use it in a cluster, since the other nodes must reach this port.
TURBINE_API_CORS_ORIGINS (api_cors_origins=) — the origins a browser may call the API from, comma-separated (https://ops.example.com). By default any origin may; since the token is sent in a header and never in a cookie, another site cannot use it on its own. Set this only if you want the browser to refuse other origins outright.

REST API¶

Every Turbine node exposes an HTTP API (default port: 8400, configurable via api_port or TURBINE_API_PORT). When a token is configured, every endpoint below except /health and /ready requires Authorization: Bearer <token> — see Securing the API.

Status & monitoring¶

Endpoint	Method	Description
`/health`	GET	Liveness + node status (always `200` while the process serves HTTP)
`/ready`	GET	Readiness: same body, `503` when stopping or unable to participate
`/ready/stop`	POST	Withdraw this node from client traffic until restart; workers, API and Raft keep running. Requires the API token when configured; returns the health body with `reason: "stopping"`.
`/assignments`	GET	Partition assignments for this node
`/metrics`	GET	Prometheus metrics; accepts the API token or the optional metrics-only token (see Monitoring)
`/worker-watermarks`	GET	Exact event-time watermarks for this node, keyed by worker ID (epoch ms); absent until event time is observed

Both health endpoints return the same body. status is healthy, degraded (a partition halted — see Error handling — a partition whose handler keeps failing while its policy drops or dead-letters the rows, the checkpoint store no longer accepting writes, or a partition whose window closing is falling behind) or unavailable (stopping, no Raft leader, not a voter, or catching up), with reason naming the cause and, in cluster mode, leader_id (null when there is no leader) and voters. A processing_errors list names each partition of the second kind — worker, topic, partition, how long it has been failing, where the rows went and the exception — see Knowing when a partition has stopped producing results. A close_backlog list names each partition whose oldest due timer is past its threshold — worker, topic, partition, due count, oldest due age and when the episode started; the outputs are late, not lost, and the entry clears itself when the backlog drains. A checkpoint block reports where state is saved, whether that destination is reachable from another node, and whether it is accepting writes right now — see Checkpoint storage. One request answers "can this node work?" — see Health probes for the probe wiring. A node holding zero partitions is healthy: with fewer partitions than nodes, that is the nominal case.

# Check node health
curl http://localhost:8400/health

# List assigned partitions
curl http://localhost:8400/assignments

# Scrape Prometheus metrics
curl http://localhost:8400/metrics

GET /cluster/nodes includes active_timers by origin (window, join, user), join_buffered_rows and join_late_rows by side (left, right), join_input_skew_ms and the exact close_frontier_ms in each worker’s statistics. Missing fields mean unavailable or not applicable, including on older nodes during a rolling update. active_windows is the rolling 10 s maximum of persistent window close timers only. Window-origin timers also include early emissions, so their timer count can exceed the pane count.

GET /worker-close-frontiers returns the local workers’ exact close frontiers as a JSON object keyed by worker ID. It uses the same API-token protection as /metrics; cluster queries fetch it from peers to preserve integer timestamps.

Exposed metrics¶

The list of metrics, how to scrape them, a Grafana dashboard and the signals to alert on are in Monitoring.

Errors¶

API operation failures use RFC 9457 Problem Details with Content-Type: application/problem+json:

{"title":"Conflict","status":409,"detail":"no other voter to transfer leadership to","code":"no_transfer_target"}

The HTTP status carries the outcome. detail explains the specific failure; code is a stable machine-readable identifier. Successful membership operations no longer return a status envelope, and refusals never return HTTP 200.

HTTP status	Meaning
`400`	Malformed JSON, missing or invalid fields, invalid query/path parameters or missing JSON content type
`401`	Missing or wrong API token
`404`	Unknown resource
`409`	Refused in the current cluster state
`413`	Request body too large
`500`	Internal failure, fatal Raft error, or worker drain failure after leaving
`502`	Leader unreachable while forwarding
`503`	Raft not running, no leader, or quorum unavailable

Membership codes include learner_not_found, voter_not_removable, empty_membership, membership_change_in_progress, no_transfer_target, no_leader, raft_unavailable, raft_stopped, leader_unreachable, quorum_unavailable, operation_timeout, raft_fatal, and leave_drain_timeout. Other errors use the status-level codes bad_request, unauthorized, not_found, method_not_allowed, conflict, payload_too_large, client_error, internal_error, bad_gateway, and service_unavailable.

The console displays detail. Reads fail over on network errors and 5xx responses and have one query retry, including the POST-based state browser. Mutations are sent once. A failed mutation may already have committed: inspect the current cluster state before retrying. curl --fail-with-body returns a nonzero exit status for refusals while retaining their explanation.

Membership commits have a 25-second wait budget, including voter removal requested by a leaving node. Adding a learner waits for it to catch up without this deadline: a large snapshot or a slow link can legitimately take longer. A write timeout (quorum_unavailable or operation_timeout) reports an unknown outcome, not a rollback. An explicitly stopped Raft node returns 503 raft_stopped; storage failures and panics return 500 raft_fatal.

The /ready probe keeps its health document when returning 503; /health always returns its health document at 200. Static console assets keep their own response format.

Cluster management (Raft mode)¶

These endpoints are available when running in Raft cluster mode. Membership changes sent to a follower are forwarded to the leader, and the follower returns the leader's response. /cluster/leave and /cluster/force_bootstrap act on the receiving node itself.

Endpoint	Method	Description	Body
`/cluster/add_learner`	POST	Add a node as Raft learner	`{"node_id": 4, "addr": "http://10.0.0.4:8400"}`
`/cluster/change_membership`	POST	Promote learners to voters	`{"add_voter_ids": [4]}`
`/cluster/leave`	POST	Make this node leave the cluster for good: its partitions are handed over, then it leaves the voters. Send it to the node that leaves.	—
`/cluster/remove_voter`	POST	Remove a voter that is already gone	`{"remove_voter_ids": [3]}`
`/cluster/remove_learner`	POST	Stop replication to permanently retired members without changing the target voting set; refuses the entire request if any id remains a voter in that set	`{"remove_learner_ids": [3]}`
`/cluster/transfer_leader`	POST	Hand the leader role to another node — the one that answered most recently	—
`/cluster/force_bootstrap`	POST	Let this node form a cluster without hearing from every peer — for a first start where a peer will never answer. `202` when accepted, `409` once the node is part of a cluster. Only when no cluster is running: a live cluster this node cannot reach gets a second one beside it	—
`/cluster/force_release`	POST	Unblock a partition whose previous owner died mid-handoff — read the procedure first, it can cost data if misused	`{"topic": "events", "partition": 3}`

/cluster/remove_learner accepts multiple ids. An already absent id is harmless, so the same request can be retried after inspecting the cluster. Membership endpoints return 200 with {} on success. Refusals use the HTTP error status and Problem Details body described above. Invalid bodies return 400, and a node without Raft returns 503. The configured API bearer token is required. See Removing a node for the retirement procedure and the limits of this operation.

When several subscriptions read the same topic, add "sub_index": 1 to the /cluster/force_release request, using the subscription index from /assignments. Omitting it is accepted only when the topic and partition identify a single subscription.

/cluster/force_release answers 200 with what it resolved ({"sub_index", "topic", "partition", "node_id", "worker_id", "epoch", "released_epoch"}), 400 when the request needs a subscription selector, 404 when the partition is not assigned to anyone, and 409 when it never changed hands (nothing to release). Any node can serve it, once it is up to date with the cluster: a node that cannot confirm it within a few seconds — just restarted, or cut off from the leader — answers 503 rather than act on an outdated view. Retry, or send the request to another node.

GET /rebalance reports what the cluster last decided about partition placement, which nodes are down but still hold their partitions, and which nodes it is currently not assigning to:

{"leader_id": 1, "last_decision_ms": 1762000000000,
 "last_nodes_changed": [2, 3], "last_partitions_moved": 4,
 "pending": {"2": {"reason": "stopped", "down_since_ms": 1762000100000,
                   "rebalance_at_ms": 1762000160000, "partitions": 3}},
 "excluded_since_ms": {"3": 1762000000000}}

pending lists the nodes that are down — stopped, or unresponsive when they went silent without stopping — and whose partitions wait for them: nobody consumes those until rebalance_at_ms, when they move unless the node is back. excluded_since_ms is the answer to "why does node 3 hold no partitions?" — it stayed down past the rebalance delay and was taken out of the pool at that time; it comes back on its own once it answers again. Only the leader knows any of this, so a node that is not the leader reports empty fields here — but /cluster/nodes carries the same block under rebalance, fetched from the leader, so that endpoint gives the same answer whichever node you ask. Its absence there means no leader could be reached, which is not the same as "nothing excluded".

/cluster/nodes also reports, per node, any partitions it cannot start because they are waiting on a handover:

{"node_id": 2, "handoff_waits": [
  {"sub_index": 1, "topic": "events", "partition": 3, "waiting_seconds": 412.5}]}

Omitted when empty. A few seconds is a normal rebalance; minutes means the previous owner died — see Recovering a stalled partition handoff.

A node whose startup was refused can also carry a refusal:

{"node_id": 2, "refusal": {"at_ms": 1793000000000, "reason": "cluster subscription declaration mismatch…"}}

The last refusal remains available through surviving nodes after that process exits, and the console shows its reason in the Nodes view. It clears after the node successfully completes admission. The field is omitted when no refusal was retained; an absent field does not prove startup succeeded. Publication is best effort: errors before the node can contact the cluster, an unavailable quorum or a hard kill can leave only the failing node's logs. A peer's refusal does not degrade the serving node's /health or /ready.

Each node's worker_stats keeps a worker's last known numbers after it stops — its partition moved to another node, or a fatal error halted it — so the entry outlives the work. Those rows say so:

{"running": false, "stopped_at_ms": 1762000000000, "messages_total": 755}

Read them as history, not as activity: the counters are frozen at their last value, so leave those rows out of anything you total per node. This endpoint totals nothing for you — every figure it returns is per worker — and summing a stopped row keeps a partition the node gave up hours ago inside its throughput, lag and DLQ numbers. The bundled console excludes them from every total and still lists the row, marked.

The opposite situation is the one worth chasing: a row without running: false whose worker has no matching entry in that node's assignments is a worker still consuming a partition the cluster has handed to someone else. Both fields are absent on a node too old to report them, where the two cases cannot be told apart in a single reading.

State introspection¶

Query the RocksDB state stores across workers for debugging, alerting dashboards, or operational visibility.

`POST /state` — per-node query¶

Query the state stores on a single node. Available in both standalone and cluster mode.

Request body:

Field	Type	Default	Description
`worker_id`	`string \\| null`	`null`	Filter to a specific worker. `null` = all workers on this node.
`prefix`	`string \\| null`	`null`	Only return entries whose key starts with this prefix. `null` = no filter.
`limit`	`int \\| null`	`100`	Maximum entries per worker. Capped at 1000.
`include_internal`	`bool \\| null`	`false`	Include `_turbine_*` keys (offsets, timer state). Raw lane only.
`lane`	`"raw" \\| "logical" \\| null`	`"raw"`	How to read the keyspace — see below.
`partition_key`	`string \\| null`	`null`	Logical lane: restrict the scan to one keying group. `""` selects the keys written without a keying scope.

Response:

{
  "workers": [
    {
      "worker_id": "work-00-p00",
      "topic": "dev-input",
      "partition": 0,
      "truncated": false,
      "entries": [
        {"key": "acc|error_rate_short|tenant_1|region_eu|1700000000000", "value": "{\"sum\":450.2,\"count\":10}"},
        {"key": "cool|error_rate_short|tenant_1|region_eu", "value": "{\"severity\":\"warning\"}"}
      ]
    }
  ]
}

truncated is true when the worker has more entries than limit. Increase limit or narrow prefix to see more.
partial is true when an unanchored logical scan hit its budget before the end of the store (see below) — the page is what was found, not what exists.
value is returned as UTF-8 text when the bytes are valid UTF-8, otherwise as b64:<base64>.
Keys starting with _turbine_ are excluded by default (these store offsets and timer metadata). Set include_internal: true to see them.

Examples:

# All state across all workers on this node
curl -X POST http://localhost:8400/state \
  -H 'Content-Type: application/json' -d '{}'

# Accumulators for a specific rule, limiting results
curl -X POST http://localhost:8400/state \
  -H 'Content-Type: application/json' \
  -d '{"prefix": "acc|error_rate_short", "limit": 50}'

# All state for one worker, including internal keys
curl -X POST http://localhost:8400/state \
  -H 'Content-Type: application/json' \
  -d '{"worker_id": "work-00-p00", "include_internal": true}'

# What this worker holds, under the names your handler used
curl -X POST http://localhost:8400/state \
  -H 'Content-Type: application/json' \
  -d '{"lane": "logical"}'

# One keying group, by name — a direct seek rather than a scan
curl -X POST http://localhost:8400/state \
  -H 'Content-Type: application/json' \
  -d '{"lane": "logical", "partition_key": "tenant_1", "prefix": "cool|"}'

`POST /cluster/state` — cluster-wide query¶

Fan-out to all Raft members, aggregate their /state responses. Only available in Raft cluster mode.

Same request body as /state. The response includes an additional unreachable field listing node addresses that did not respond within 5 seconds, and an unauthorized field listing nodes that refused this node's API token (both omitted when empty). Cluster-wide list endpoints return the available data with these fields, even when that data is empty. Check them before treating an empty list as complete. Point reads use 503 when they cannot determine absence, as described below.

Response:

{
  "workers": [
    {"worker_id": "work-00-p00", "topic": "dev-input", "partition": 0, "truncated": false, "entries": [...]},
    {"worker_id": "work-01-p00", "topic": "dev-input", "partition": 1, "truncated": true, "entries": [...]}
  ],
  "unreachable": ["10.0.1.3:8400"]
}

Example:

# All aggregate state across the entire cluster
curl -X POST http://any-node:8400/cluster/state \
  -H 'Content-Type: application/json' \
  -d '{"prefix": "agg|"}'

What `prefix` matches — and what it does not¶

prefix is matched against the key as stored, and a stored key is not the key your code wrote. Turbine prepends a small binary header that routes the key to its shard, so a key your handler put under cool|{rule}|{group} lands on disk behind a few non-printable bytes. In the default raw lane a prefix of cool| matches nothing — not because the data is missing, but because the match starts one header too early.

lane picks which reading you want:

`lane`	`prefix` matches	Keys returned	Reaches
`"raw"` (default)	the stored bytes, header included	verbatim	everything, internal keys included
`"logical"`	the key your handler wrote	decoded, with the keying group they belong to	every user key; no internal keys

In the logical lane each entry also carries partition_key (the keying value the key is stored under, absent when the write used no keying scope) and key_group (the shard the header routes it to):

{"key": "cool|error_rate|tenant_1", "partition_key": "tenant_1", "key_group": 742,
 "value": "{\"severity\":\"warning\"}"}

partition_key is the difference between a seek and a walk. With it, the header is known, so the query goes straight to that group: a lookup, whatever the store holds. Without it there is nothing to seek to, so the scan walks the store and filters — bounded both by a number of keys and by a deadline, and reporting partial: true when it gave up before the end. An empty page from a partial scan means "not found so far", not "not there".

What that costs, in practice, on a worker holding a million keys: the default view and any prefix that matches often stop as soon as the page is full, so they are sub-millisecond; a prefix that matches nothing is the one that walks, and it is the case the deadline exists for. Anchor the search on a keying group whenever you know it — it is the difference between a lookup and a search, and it stays a lookup as the store grows.

include_internal is rejected in the logical lane rather than quietly returning nothing: internal keys carry no header, so they have no decoded form and are only reachable as {"lane": "raw", "prefix": "_turbine_"}.

Key	Prefix that matches (raw lane)	Content
`_turbine_offset`	`_turbine_`	Last committed offset (internal)
`_turbine_timer:{id}`	`_turbine_timer:`	Scheduled timer fire times (internal)

To read an aggregate — one window, one group, values rather than stored bytes — use the windows endpoints instead. They rebuild the key layout for you and hand back what the accumulator computes.

`POST /cluster/state/key` — one key, decoded¶

Read back one value your handler wrote, decoded, from whichever node holds it — the lookup an application's own API makes ("the risk score of user u1"), where /state is a browse. POST /state/key does the same on this node only.

Field	Type	Description
`subscription`	`string`	The subscription's name: the handler's name, or `@subscribe(name=...)`.
`key`	`string`	The key as the handler wrote it (`state.put(key, ...)`).
`partition_key`	`string \\| null`	The scope it was written under (`state.scope(partition_key)`). Omit it for an unscoped write.
`topic`	`string \\| null`	Only when two subscriptions with that name read different topics.
`worker_id`	`string \\| null`	Restrict the read to one worker.

curl -X POST http://localhost:8400/cluster/state/key \
  -H 'Content-Type: application/json' \
  -d '{"subscription": "score_users", "key": "risk", "partition_key": "u1"}'

{"subscription": "score_users", "topic": "events", "key": "risk",
 "partition_key": "u1", "worker_id": "work-00-p03",
 "value": {"score": 7, "level": "high"}, "decoded": true}

value is decoded with the app's state_format. A value that does not decode (raw bytes written with put_bytes) comes back as UTF-8 text, or base64 with a b64: prefix, and decoded is false.

Where it looks: the node that received the request first, then the node that owns the partition Kafka's default partitioner sends that partition_key to, then — only if neither has it — every other node at once. That last step keeps the answer right when your producer places keys differently, or while partitions are moving; it only costs time. A key written without a scope is held separately by each partition, so with several partitions the first one found answers; use worker_id to pick one.

A missing key is 404 only after every queried node could search its recovered state and answered without finding it. If no value is found and any node could not answer, the response is 503 with Content-Type: application/problem+json:

{"title": "Service Unavailable", "status": 503, "detail": "cluster read incomplete: ...", "code": "service_unavailable", "unreachable": ["http://turbine-0:8400"], "unauthorized": []}

Both lists are always present on a 503: unreachable includes connection failures, timeouts, server errors and invalid responses; unauthorized lists nodes that refused the API token. Keep the last value and retry after an unavailable read instead of clearing it. A value found on any queried node still returns 200. A local store read failure answers 500 with the reason; a peer's store failure makes that peer unreachable and an otherwise missing read 503.

HTTP can answer before the application has finished admission or its workers have recovered. Point reads also return 503 during that interval. A value found on a ready worker remains usable while other workers recover; a miss waits for every relevant worker. worker_id narrows which workers must be ready. During a partition transfer, a node whose assignment view has not caught up, or a cluster search that crosses a placement change, also answers 503. Retry after recovery or placement has settled; an HTTP response alone does not prove that a node can confirm absence.

A 503 can persist until an operator intervenes: a halted partition or an EOS worker waiting at the handoff barrier cannot certify absence. Check worker health and the handoff status before deciding whether to restart a halted worker or resolve a blocked handoff. Successful local reads do not wait for a Raft leader or quorum; only confirming absence requires synchronization.

Windows¶

Read what your windows are doing: which ones exist, which groups have one open right now, and — the question an alert usually sends you to ask — what the numbers behind an aggregate actually were.

Endpoint	Method	Description
`/windows`	GET	Windows declared on this node: name, label, kind, keys, size / slide / gap, accumulator. Paged — `?q=` searches name or label, `?limit=` sizes (default 250, max 1000)
`/windows/calculations`	GET	Observed calculation hosts for one exact `name`, paged by worker
`/windows/{name}/entries`	GET	Open windows for one name: one row per `(group, window start)`, with the time it fires
`/windows/{name}/entry`	GET	One open window's stored state, decoded
`/windows/{name}/series`	GET	One group's series: a value per pane (or per open window), plus the aggregate they roll up to
`/cluster/windows`, `/cluster/windows/calculations`, `/cluster/windows/{name}/entries`, `/cluster/windows/{name}/entry`, `/cluster/windows/{name}/series`	GET	Same, fanned out across the cluster

Windows created inside process() (one per rule, per tenant, …) appear here as soon as they are constructed — nothing has to be declared up front. That is also why the list is paged: an app driven by configuration can declare thousands, and this endpoint is polled every few seconds by every open console, from a node that also asks each of its peers. The response carries total and truncated so a page never reads as the whole set.

Search server-side, not on the page you got. ?q= is matched before the page is cut, so it reaches windows the page never included; filtering the response instead would only ever find what the cut happened to keep. The console does this for you — its search box drives q.

# Which rules watch latency, whatever their id?
curl 'http://localhost:8400/cluster/windows?q=latency'

On the cluster endpoint, total is a lower bound when truncated: a peer that cut its own page did not name what it left out.

Each open window reports the time it fires, and which clock that time is on: an event-time window closes when its partition's watermark reaches the deadline, not when wall clock does, and those entries carry fire_time_is_event_time: true. The distinction is the difference between a useful countdown and a confident wrong one — replaying history, every event-time window would otherwise read as long overdue. Measure the wait against the owning worker's close_frontier_ms in /cluster/nodes: this is the engine's sampled closing progress, including idle coasting. The console uses it for the relative label and progress bar and shows the data watermark separately. The data watermark (/cluster/nodes) tracks consumed event time and can stay fixed while coasting closes a quiet partition's windows.

Closing progress updates on refresh; the browser does not predict coasting. An absent close_frontier_ms means closing progress is unavailable (for example, on an older node or before a closing bound is known). The console then keeps the absolute deadline and omits the countdown and progress bar. It labels stopped or missing owning workers separately from missing closing diagnostics.

This applies to both list and detail responses. A processing-time window still uses wall clock even when its subscription publishes a Kafka watermark.

Calculation hosts describe the constructed operator. Native fixed-window calculations report "rust". Python fallbacks and session calculations report "python". Placement, timers and closing run in Rust for both hosts.

Add calculation_summary=true to either list endpoint to receive calculation_counts with rust and python counts instead of calculations. The console uses this option for polling. The cluster adds the observations reported by each responding node. If multiple nodes observe the same worker and host, each node's observation contributes to the count. Zero means no observation was reported for that host. Registration can be lazy, and older nodes can omit the diagnostic entirely. Observations are removed when workers stop and are collected afresh after restart. Calls without the option still return the existing per-worker calculations array when observations exist.

Open Calculation workers in the console, or request a page directly.

curl --get 'http://localhost:8400/cluster/windows/calculations' \
  --data-urlencode 'name=alert-rule:17761784b6' --data-urlencode 'limit=50'

Each page returns calculations sorted by worker ID and host. The cluster deduplicates identical pairs and keeps both rows when a worker is observed with both hosts. limit counts worker IDs (default 50, range 1–250). Pass next_after back as after for the next page, or use worker_id to inspect one worker. An empty array means no matching observations were returned. Pages reflect live registrations at the time of each request. Cluster responses report missing and unauthorized peers as partial data. A peer running an older version can contribute counts and worker observations via its legacy list. The detail endpoint looks for the exact name in the first 50 search matches on that peer. If the search was truncated before reaching the window, incomplete lists that peer.

Naming a window without renaming its state¶

A window's name is its identity: it prefixes every state and timer key it owns. Renaming it abandons the panes, timers and partial aggregates already on disk — so an application that creates windows from configuration has to name them after something stable, typically an id, which is exactly what nobody can read.

label= is the way out. It is display metadata, never part of a key:

win.PersistentSliding(
    self,
    name=f"alert-rule:{rule.id}",     # identity — stable, never changes
    label=rule.title,                  # what a human reads
    size_ms=600_000,
    slide_ms=60_000,
    key=["tenant_id", "runner_id"],
    value=agg.Mean("latency_ms"),
    on_close_each=emit,
)

Resident fixed windows expose storage: "resident" in their metadata. Their inventory, detail and series endpoints read committed live columns with the same group filters and limits; bounded results report truncated. Checkpoint/replay durability and supported features are described in Windowing.

/windows returns it alongside the name, and the console titles the row with it (keeping the name visible — it is what logs and alert payloads carry). Call window.set_label(new_title) when your configuration changes at runtime: the label updates without touching a single key, so nothing is lost and no restart is needed.

Reading the values¶

Window starts, ends and close deadlines are signed 64-bit Unix milliseconds. Negative values represent instants before 1970; zero is a valid instant. The ws parameter on /entry accepts the same signed window start returned by /entries. Ends that would exceed Int64 saturate at its upper limit. Durations such as size_ms, slide_ms and allowed lateness remain nonnegative. Persistent close deadlines have a lower bound of zero: an event-time deadline before 1970 is reported as 1970-01-01T00:00:00Z, while the window's start and end retain their signed values.

# Which groups have a window open?
curl 'http://localhost:8400/windows/alert-rule:17761784b6/entries?limit=20'

# The curve for one of them.
curl 'http://localhost:8400/windows/alert-rule:17761784b6/series?tenant_id=acme&runner_id=r-12'

{
  "name": "alert-rule:17761784b6",
  "label": "Net latency per runner",
  "kind": "sliding",
  "group": {"tenant_id": "acme", "runner_id": "r-12"},
  "worker_id": "work-00-p00",
  "unit": "pane",
  "size_ms": 600000,
  "slide_ms": 60000,
  "points": [
    {"start_ms": 1788464400000, "end_ms": 1788464460000, "value": 41.2,
     "state": {"latency_ms_sum": 206.0, "latency_ms_count": 5}},
    {"start_ms": 1788464460000, "end_ms": 1788464520000, "value": 58.7,
     "state": {"latency_ms_sum": 176.1, "latency_ms_count": 3}}
  ],
  "aggregate": 47.8,
  "aggregate_window_start_ms": 1788463920000,
  "finalized": true,
  "truncated": false
}

Every key the window declares must be given as a query parameter: a series is per group, and a half-specified group would blend two curves into one. A group with nothing open answers 404. For the cluster-wide /series and /entry reads, this requires every queried node to have answered. If nothing is found and any peer could not answer, they return 503 Problem Details with detail, unreachable and unauthorized, using the same error body as key reads. Keep the last series or entry and retry; an unavailable owner does not mean the group's state disappeared. Local storage failures, aborted reads and in-memory snapshot failures answer 500. When a peer encounters one and no other node supplies the value, the cluster read reports that peer as unreachable in its 503 response.

What the fields mean:

value is what the accumulator computes — a mean is a number. It is finalized for you, so nothing on the reading side needs to know that a mean is persisted as a sum and a count, or that a t-digest is opaque bytes.
state is the raw record behind it, in case you are debugging the window rather than reading it.
unit says how to read the points. pane — a sliding window's points are slices of one result, so they form a curve and roll up into aggregate (the value the window would emit if it closed now, covering everything from aggregate_window_start_ms — which marks that span whether or not a value could be computed). window — a tumbling or session window's points are each a result of their own, so there is no aggregate: merging them would produce a number no window ever emits.
finalized: false means no accumulator answered on that node, so only state is populated. It distinguishes "this window computed nothing" from "this node cannot tell you what it computed".

The series covers what is still open: a sliding window's held panes (the last size_ms), or the windows a group has not closed yet. Closed windows are gone from state by design — their result was emitted downstream, and that output stream is where history lives.

Two bounds worth knowing: at most 1000 points come back per read (?limit= lowers it, truncated says when it bit), and ?worker_id= restricts the read to one worker when you already know which partition owns the group.

Topology¶

GET /topology returns the application's static topology: one entry per subscription with its consumed / produced / dead-letter topics, error and delivery policies, and declared windows. In a cluster it combines the nodes' declarations, including subscriptions deployed to only some nodes during a rolling update. An unreachable or unauthorized peer is reported in the corresponding response field; the console marks this as partial data. For a shared subscription whose configuration differs during rollout, the queried node's declaration takes precedence. Live numbers (rates, lag, DLQ counts) come from /cluster/nodes; Join them through subscription identity (sub_index), since several subscriptions may read the same topic.

curl http://localhost:8400/topology

For an interval join, the subscription also contains interval_join: its on columns, inclusive between bounds, how, emit, dedupe_left, allowed_lateness_ms and expiry_bucket_ms. inputs lists the left and right topics, and the console draws both edges into that subscription.

{
  "subscriptions": [
    {
      "sub_index": 0,
      "name": "score_cpu_usage_per_user",
      "source_topic": "events",
      "output_topic": "scores",
      "dlq_topic": "events-dlq",
      "on_error": "dlq",
      "on_processing_error": "fail",
      "late_topic": "events-late",
      "on_late": "route",
      "processing_guarantee": "exactly_once",
      "partition_key": "tenant_id",
      "event_time": true,
      "windows": ["risk_score"]
    }
  ]
}

name is the handler's function or class name (or an explicit @app.subscribe(name=…)). windows lists statically declared windows only — windows created lazily inside process() appear in /cluster/windows once the app is running, but not here. Optional fields (output_topic, dlq_topic, late_topic, partition_key, empty windows) are omitted.

A subscription's outgoing topics are output_topic (results), dlq_topic (undecodable or failed records — see error handling), late_topic (rows past their window's lateness budget, when on_late is "route" — see windowing), and any number of side_outputs — the named destinations the subscription declared, whether written imperatively with ctx.emit or by a branch= arm (see side outputs). on_error and on_late are always present, including under their default policies. on_processing_error appears only when it differs from on_error — absent means the handler phase follows it, which is the common case; present is the fact a client has to show, because on_error alone would then describe the wrong outcome for the phase that is actually failing. side_outputs is omitted when empty and otherwise looks like:

"side_outputs": [
  { "name": "critical", "topic": "alerts-critical", "predicate": "score >= 90" },
  { "name": "audit", "topic": "audit-log" }
]

predicate is the rule a declarative branch= arm applies, rendered as text. It is absent on a destination written with ctx.emit, and that absence is information rather than a gap: an imperative emit's rule lives in the handler's Python and cannot be reported. Do not treat a missing predicate as "matches everything".

A side output's name is local to its subscription, so join on (sub_index, name) — two subscriptions may use the same name for different things, and one subscription may point two destinations at one topic. The order of the array is the order the destinations were declared, which for branch= arms is also the order they are evaluated in: the first arm whose predicate holds takes the row.

sub_index is the subscription's declaration order, and matches the sub_index on /assignments. Since several subscriptions may read one topic, a topic can appear on several rows — join a subscription to its workers on sub_index, never on source_topic, or you will merge the stats of every subscription reading that topic.

Events¶

Each node keeps a small in-memory ring (512 entries) of operational events: worker starts, stops, failures, restarts. This is what makes a crash-looping worker visible — gauges show the same value between two polls, the event feed shows every restart in between.

Endpoint	Method	Description
`/events`	GET	This node's events, oldest first. `?since=<seq>` cursors on the per-node sequence, `?limit=` caps (default 200, max 512)
`/cluster/events`	GET	Every node's feed merged, newest first, each event stamped with its reporting `node_id`; identified startup refusals deduplicated before the limit

curl 'http://localhost:8400/cluster/events?limit=50'

{
  "events": [
    {
      "seq": 42,
      "timestamp_ms": 1781278835645,
      "kind": "worker_failed",
      "worker_id": "work-00-p00",
      "topic": "events",
      "partition": 0,
      "node_id": 1,
      "message": "state store error: ..."
    }
  ]
}

kind is one of worker_started, worker_stopped, worker_failed, worker_fatal (the error that stops one partition for good), worker_restarted, force_release (someone forced a stalled handoff — a manual action, recorded so it stays auditable), node_refused (a node could not start; its message names that node and gives the reason), rebalance (the cluster moved partitions), voter_down (a node stopped answering but keeps its partitions during the rebalance delay), voter_recovered (it answered again before exclusion; the message gives its downtime), and voter_excluded / voter_included (it was taken out of the assignment pool, or came back after exclusion).

Startup refusals carry subject_node_id (the node that could not start) and an opaque event_id, stable across nodes and node restarts. node_id continues to identify the reporting node, and seq its local cursor. A refusal can appear in several local /events feeds; /cluster/events returns one copy per event_id before applying its limit. Different refused starts remain separate even when their reasons and timestamps match. The console shows the refused node and names the reporter in a tooltip.

Older nodes omit the two optional fields. Their events remain visible, potentially with duplicate refusals in a mixed-version cluster; consumers must fall back to (node_id, seq) rather than guessing identity from message text or timestamps. Use event_id for an identified refusal's display identity across cluster polls; continue to use the reporting node's seq for local /events?since= polling.

The ring is in-memory: it resets on process restart, and long-horizon history belongs to your logging stack — this feed answers "what just happened?", not "what happened last week?".

Dead-letter queue (DLQ) inspection¶

Surface which subscriptions route bad records where, how many have been dead-lettered, and what the dead-lettered records actually look like — without dropping to rpk. See error handling for the on_error / dlq_topic configuration these endpoints report on.

Endpoint	Method	Description
`/dlq`	GET	This node's subscriptions with their DLQ config + counters
`/cluster/dlq`	GET	Same, aggregated across the cluster (Raft mode)
`/dlq/peek`	GET	The last N raw records off a dead-letter topic

A lightweight recent-DLQ count is also folded into each worker on /cluster/nodes (dlq_60m, the rolling 60-minute count) so a node card shows DLQ activity at a glance.

`GET /dlq` and `GET /cluster/dlq`¶

curl http://localhost:8400/dlq

{
  "subscriptions": [
    {
      "name": "process_orders",
      "source_topic": "orders",
      "dlq_topic": "orders_dlq",
      "on_error": "dlq",
      "on_processing_error": "fail",
      "counters": {
        "total": 128,
        "recent_60m": 12,
        "by_phase_action": [
          {"phase": "deserialize", "action": "dlq", "count": 120},
          {"phase": "processing", "action": "dlq", "count": 8}
        ]
      }
    }
  ]
}

Every subscription is listed with its configured on_error policy, so a subscription using skip (no dlq_topic) still shows up with its drop counters. on_processing_error is served alongside it only when the two differ — a subscription that dead-letters bad bytes but halts on a handler exception does not do what an on_error: "dlq" row alone suggests. total is cumulative since process start; recent_60m is the rolling 60-minute count. /cluster/dlq sums each subscription's counters across nodes and adds an unreachable list for any peer that did not respond.

A subscription is identified by (source_topic, name), since several subscriptions may read one topic.

Drop counters belong to each subscription independently, even when several subscriptions read the same topic. They include all of a subscription's partitions and, for multi-input subscriptions, all input topics. A subscription with no drops reports zero counters. The cumulative Prometheus counter retains the actual source topic; the rolling 60-minute gauge counts all inputs once per worker and uses the subscription's anchor topic as its topic label.

`GET /dlq/peek`¶

curl 'http://localhost:8400/dlq/peek?topic=orders_dlq&limit=10'

topic must be a configured dlq_topic (the endpoint refuses arbitrary topics). limit defaults to 20 (max 200). The read is non-destructive — it never commits an offset.

{
  "topic": "orders_dlq",
  "records": [
    {
      "partition": 0,
      "offset": 41,
      "timestamp_ms": 1781190000000,
      "key": "tenant_42",
      "payload": "{not valid json",
      "payload_truncated": false,
      "headers": [
        {"key": "turbine_dlq_source_topic", "value": "orders"},
        {"key": "turbine_dlq_source_partition", "value": "0"},
        {"key": "turbine_dlq_source_offset", "value": "41"},
        {"key": "turbine_dlq_phase", "value": "deserialize"},
        {"key": "turbine_dlq_error", "value": "Json error: ..."},
        {"key": "turbine_dlq_worker", "value": "work-00-p00"}
      ]
    }
  ]
}

Keys, payloads, and header values are UTF-8 text where possible, otherwise base64 with a b64: prefix. Payloads over 2 KB are truncated (payload_truncated: true). Replay (re-producing from the DLQ back to the source topic) is out of scope — peek + counts only.