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. Build it from a clone of the repository (just console-build) and serve console/dist/ 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, and the order matters: the console tries them top to bottom and stops at the first that answers. A node that is down or broken is skipped and the next one is tried, so listing several nodes buys you a console that survives losing any one of them. 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.

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.

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 windows and their contents, split into Persistent (tumbling, sliding, session) and In-Memory. 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?".
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.

Before exposing it

The API has no authentication and accepts requests from any origin. 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. Treat the API port as an internal admin interface: keep it on a private network or behind your own authenticating proxy, and never publish it to the internet. This is a property of the API itself, not of the console — the same applies to the curl calls below.

REST API

Every Turbine node exposes an HTTP API (default port: 8400, configurable via api_port or TURBINE_API_PORT).

Status & monitoring

Endpoint	Method	Description
/health	GET	Liveness + node status (always 200 while the process serves HTTP)
/ready	GET	Readiness: same body, 503 when the node cannot work
/assignments	GET	Partition assignments for this node
/metrics	GET	Prometheus metrics (see below)

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, or the checkpoint store no longer accepting writes) or unavailable (no Raft leader, or this node is not a voter), 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 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

Exposed metrics

Metric	Type	Description
turbine_messages_total	counter	Total messages processed
turbine_batches_total	counter	Total batches processed
turbine_throughput_10s	gauge	Messages/second over the last 10s
turbine_throughput_60s	gauge	Messages/second over the last 60s
turbine_state_ops_total	counter	Total state operations committed (user puts/deletes + offset)
turbine_state_ops_10s	gauge	State operations/second over the last 10s
turbine_state_ops_60s	gauge	State operations/second over the last 60s
turbine_state_keys_expired_total	gauge	State keys removed by state_ttl since the worker started (label: worker). Reported as a gauge because the removals are counted by the storage engine's own background threads and read back as a running total
turbine_state_expiry_frontier_ms	gauge	How far state expiry has advanced, epoch ms (state_ttl subscriptions only)
turbine_consumer_lag	gauge	Consumer lag (messages behind high watermark)
turbine_event_time_watermark_ms	gauge	Event-time watermark per partition, epoch ms (event-time subscriptions only)
turbine_event_time_watermark_lag_ms	gauge	Wall clock − watermark, refreshed ~10 s — keeps growing on a silent stream
turbine_event_time_null_total	counter	Rows with missing/unparsable event time, dropped from event-time bucketing
turbine_dlq_messages_total	counter	Dead-lettered / dropped records (labels: topic, partition, phase, action)
turbine_dlq_messages_60m	gauge	Dead-lettered / dropped records over the last 60 min
turbine_last_checkpoint_timestamp_ms	gauge	Wall-clock time of the last uploaded checkpoint
turbine_last_checkpoint_offset	gauge	Consumer offset captured by that checkpoint — a recovery replays everything after it
turbine_last_checkpoint_size_bytes	gauge	On-disk size of that checkpoint
turbine_checkpoint_store_shared	gauge	0 when the checkpoint destination is local to this node. Alert on it together with cluster mode: a stateful cluster on a node-local store loses state on every partition handoff
turbine_checkpoint_store_writable	gauge	0 when the store's last answer was a failure. Kept current by real uploads, and by a small probe when uploads go quiet (an idle partition never checkpoints)
turbine_checkpoint_consecutive_failures	gauge	Store failures since the last success. Past three, /health reports degraded
turbine_checkpoint_failures_total	counter	Cumulative checkpoint failures by step (operation: upload, prune, probe, snapshot)
turbine_poll_seconds	histogram	Time waiting for messages from the broker
turbine_decode_seconds	histogram	JSON → Arrow deserialization time
turbine_process_seconds	histogram	User business logic time
turbine_produce_seconds	histogram	Encoding + sending to the output topic
turbine_commit_seconds	histogram	Writing state + offset to RocksDB
turbine_batch_duration_seconds	histogram	Total batch time (all phases)
turbine_worker_kind	gauge (info)	Constant 1 carrying the worker's nature as a kind label: stateful, stateless or repartition. Join it onto any worker-labeled series with PromQL group_left(kind) to slice dashboards by worker nature
turbine_worker_running	gauge	1 while the node runs this worker, 0 once it stops (partition revoked, or halted by a fatal error). Per-worker series are never removed from a scrape — a counter cannot be — so this is what separates a worker's last values from its current ones
turbine_worker_stopped_at_ms	gauge	Wall-clock time the worker stopped, 0 while it runs. Answers "since when": a worker that stopped seconds ago is a rebalance in flight, one that stopped an hour ago is history
turbine_worker_restarts_total	counter	Supervisor restarts of a worker. A steadily growing value is the crash-loop signature — alert on it: throughput at zero alone cannot distinguish a crash-loop from an idle topic
turbine_worker_fatal_total	counter	Fatal worker halts, cumulative per worker. The halt is contained to that partition — the app keeps running
turbine_halted_workers	gauge	Partitions currently stopped by a fatal halt. Alert on > 0: throughput on the remaining partitions hides the hole otherwise
turbine_processing_errors_consecutive	gauge	Batches this worker's handler has failed in a row while skip / dlq kept it running; back to 0 on the first one that goes through. Alert on it: nothing else says the partition has stopped producing results — it is not halted and its offsets keep advancing. Past three, /health reports degraded
turbine_raft_has_leader	gauge	1 when this node sees a Raft leader, 0 when it does not (no quorum → nothing can be assigned or rebalanced). Cluster mode only
turbine_liveness_excluded_nodes	gauge	Cluster nodes currently left out of partition assignment because they stopped answering. Only the leader can measure this; every other node reports 0, so read it as max() across nodes. Cluster mode only
turbine_handoff_barrier_waiting	gauge	1 while a partition (labels: topic, partition) waits for its previous owner to hand over. Cluster + exactly-once only
turbine_handoff_barrier_waiting_seconds	gauge	How long that wait has lasted. Seconds is normal during a rebalance; minutes means it is stuck and needs manual recovery
turbine_repartition_records_total	counter	Records re-emitted by a repartition hop, labeled by destination partition — compare partitions to spot key skew
turbine_repartition_null_keys_total	counter	Repartitioned rows whose grouping column was null (routed to the shared empty-key group)
turbine_repartition_transit_ms	histogram	Repartition hop transit: re-emit time minus source record timestamp. Near zero when the hop keeps up; grows when it falls behind

Histograms are exposed as summaries with per-worker quantile="0.5|0.9|0.99|…" lines plus _sum/_count. The quantiles are computed per worker over a sliding window — read them per worker (take a max for a "worst worker" view), never sum or average them across workers.

All metrics are labeled with worker="work-{sub_index}-p{partition}", where sub_index is the subscription's declaration order. A worker belongs to exactly one subscription, so when several subscriptions read one topic they have distinct worker labels — group by worker (or join on the sub_index from /assignments) rather than by topic to keep them apart.

Cluster management (Raft mode)

These endpoints are available when running in Raft cluster mode. They support ForwardToLeader: if you hit a follower, you get a redirect response with the leader's address.

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/remove_voter	POST	Remove a voter (drain before maintenance)	{"remove_voter_ids": [3]}
/cluster/transfer_leader	POST	Hand the leader role to another node	—
/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/force_release answers 200 with what it resolved ({"topic", "partition", "node_id", "worker_id", "epoch", "released_epoch"}), 404 when the partition is not assigned to anyone, and 409 when it never changed hands (nothing to release). Any node can serve it.

GET /rebalance reports what the cluster last decided about partition placement, 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,
 "excluded_since_ms": {"3": 1762000000000}}

excluded_since_ms is the answer to "why does node 3 hold no partitions?" — it stopped answering 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": [
  {"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.

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.

Response format:

// Success
{"status": "Ok"}

// Redirect (hit a follower)
{"status": "ForwardToLeader", "leader_id": 1, "leader_addr": "http://10.0.0.1:8400"}

// Error
{"status": "Error", "message": "..."}

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.

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.

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, ?kind=persistent\|in_memory scopes, ?limit= sizes (default 250, max 1000)
/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/{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 worker's watermark (/cluster/nodes, or the console, which does it for you).

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,
)

/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

# 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 — distinct from an empty 200, which would read as "this group has no data" rather than "check the key".

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. It is a pure function of the app's @subscribe declarations, so it is identical on every node — query any one of them. Live numbers (rates, lag, DLQ counts) come from /cluster/nodes; join the two by topic.

curl http://localhost:8400/topology

{
  "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 node_id

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), rebalance (the cluster moved partitions), voter_excluded / voter_included (a node stopped answering and was taken out of the assignment pool, or came back). 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 are per topic, not per subscription

Dead-letter counters are recorded against the source topic. When several subscriptions read the same topic they each report that topic's counters, so the same drops appear on each of those rows — don't add them up. The on_error / on_processing_error / dlq_topic config on each row is per subscription and exact.

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.