feat(manager): deliver events across instances and report live progress

[ekalinin]

Stacked on #15.

Watch across instances (I2)

Watchers lived in a per-process map and Redis Pub/Sub carried only STOP_QUERY. A subscription opened through any instance other than the query's owner never received an event, which breaks I2 for both WebSocket and WatchQuery.

Events now travel the same control channel as QUERY_EVENT and are fanned out to the local subscriptions of every instance. An instance ignores its own announcements, otherwise every event would be delivered twice. Publishing runs in its own goroutine behind a bounded queue: it must never stall the query that produced the event, so an overflow is dropped with a log line - readers can always poll the record.

Progress

Stats were written once, at completion, so a query that ran for an hour reported nothing until it was already over. The encoder calls back every thousand rows, at most once a second otherwise, and once more when the stream ends - a row-count trigger alone would have missed exactly the case this exists for, a long aggregation that returns a handful of rows. The manager persists and announces the row and byte counts from there, throttled to one write every couple of seconds, with the first batch reported immediately so a client sees the query moving.

The persist is conditional on the record still being RUNNING, and reports stop once it is not. An unconditional PutQuery walked straight through the fence failOwnerLost puts up: a record another instance's reaper had already failed went back to RUNNING with its error and FinishedAt cleared, rejoined this instance's in-flight set with no lease behind it, and got failed again on the next tick - a FAILED <-> RUNNING ping-pong announcing a false terminal state to the cluster every couple of seconds.

It also runs off the row loop, behind a single-slot mailbox. Inline, every report parked the loop for as long as the MetaStore took to answer - up to the full persist timeout on a degraded Redis, six times a minute, on a context that could not be canceled.

Event delivery

A terminal event is what closes a remote subscription, and nothing follows it, so it is no longer dropped when the publish queue is full: dropping it left every WebSocket connection and Connect stream held by another instance open for good. The publisher is also stopped after the query goroutines it publishes for, and by way of its own queue rather than the shared context, so the last queries' terminal events still get out during a shutdown.

In the other direction, a remote event for a query this instance is executing is ignored - the owner reports on itself. Delivering another instance's verdict closed the local subscriptions and made a mode: sync submission return FAILED for a query that was still running and went on to succeed.

STOPPABLE

The spec defines SERVING -> DRAINING -> STOPPABLE, but the third state was unreachable. A draining instance advances to it once it holds zero owned in-flight queries; /v1/admin/can-stop and CanIBeStopped report it, and /readyz keeps returning 503 through it, so the node stays out of rotation until termination.

Telemetry

There were two parallel metric stacks: the domain metrics on prometheus/client_golang, and beside them an OTel MeterProvider that never received a single instrument, so nothing reached OTLP.

There is one source of truth now: the instruments are OTel and Prometheus is one of the readers behind them. Names, labels and histogram buckets are preserved, so existing dashboards and alerts keep working, while OTLP finally carries the same data. Go runtime metrics stay on the Prometheus collector, which is what exposes the full runtime/metrics ruleset.

ensureMeterProvider now refuses a resource or an OTLP endpoint that arrives after the provider was already built, instead of dropping them. Handler() builds one lazily, so serving /metrics before InitOTel left OTLP exporting nothing for the life of the process, with target_info missing its service.name and no diagnostic anywhere. The OTLP reader is also built before the Prometheus exporter, which registers a collector with the default registry as a side effect and offers no way to remove it - failing after that call left the collector behind, and the next attempt registered a second one.

Execution spans also lost their parent, because I1 detaches the execution context from the request and a child span is therefore impossible. The submitting span context is carried over as a trace.Link - the relationship is recorded without pretending the two share a lifetime.

Contract changes

CanIBeStoppedResponse.instance_state and QueryRecord.subject are added to the proto (regenerated with buf generate), to OpenAPI and to the REST DTOs.

Tests

A watch opened on a non-owner instance over a shared MetaStore receives the terminal event; progress is observed before the query finishes; the transition to STOPPABLE and the fact that it still reads as draining for readiness; domain metrics appear on /metrics with their labels.

Two regressions were added for the fixes above and were both checked against the pre-fix code, where they fail: progress does not put a fenced record back to RUNNING, and a remote terminal event does not close the watchers of a query this instance is still executing. Plus one for the late-configuration case in telemetry.

Verification

go test -race ./..., golangci-lint run ./..., buf lint - clean.