feat(indexer): drain indexing pipelines on planned teardowns

[Totodore]

What

Adds a generic pipeline drain mechanism so planned teardowns (plan-driven pipeline shutdowns, node shutdown) publish and settle in-flight batches instead of dropping them.

Why

Some sources dont support partitionned consumers or dont support checkpoints-based consumers. These sources need a way to acknowledge the incoming messages once they are indexed even when shutting-down a pipeline.
Without a graceful source drain, these ack-only sources are subjects to at-leat-once delivery whereas they could be in the exactly-once delivery scenario (except in case in of crashes).

Our specific nats case

We want to implement a Nats source. Nats does not support partitionned consumers, so there is two ways to make it work:

• Use a Nats Ordered consumer with an AckPolicy set to "None" and use the quickwit Checkpoint mechanism. However this does not allow for pipeline scaling (because without named partitions we cant key checkpoints per-pipelines). Scaling is required for our nats implementation.
• Skip the checkpoint mechanism and ack messages in suggest_truncate, with this solution it is possible to scale the number of pipelines as Nats dynamically spread load on every consumer instance (each pipelines). The tweak is that we must be able to keep the exactly-once delivery semantic in case of pipeline teardown, hence the drain mechanism.

Other source could benefit from that, the GCP pubsub source has a lot of todo related to these issues.

Design

Mirrors the FinishPendingMergesAndShutdownPipeline pattern: everything stays inside the actor system, no detached tasks, no caller-side timeouts.

• Opt-in gate: sources opt into draining through Source::should_be_drained() (default false), captured once at spawn. Pipelines whose source does not opt in are torn down by the DrainPipeline handler exactly as a kill would (terminate the child actors, exit) — sources other than the future acknowledgment-based ones are guaranteed to keep their pre-existing teardown semantics.
• Drain (SourceActor): stops emit_batches, pushes an empty force-commit batch so the indexer flushes its workbench, and the source exits with success on its own once everything it delivered is settled (Source::is_drained).
• DrainPipeline (IndexingPipeline / MetricsPipeline): fire-and-forget initiation. The shared DrainState records the pipeline's drain deadline (own commit timeout + 30s grace) and guards against respawn; the pipeline's existing supervise loop enforces the deadline, and the source's success exit cascades until the pipeline exits with success by itself.
• IndexingService: shutdown_pipelines sends DrainPipeline to every detached pipeline without awaiting; the supervise loop reaps the handles by actor state. Merge pipelines survive their draining indexing pipelines so the final force-committed splits still reach the merge planner (immature-split pickup is node-scoped).
• DrainAllPipelines (node shutdown, before universe.quit()): deferred reply completed by the supervise loop once every draining pipeline — including ones a plan change already detached — has exited. Ordered after the ingester decommission, which needs pipelines still consuming, and bounded by the new indexer.shutdown_drain_timeout config (QW_INDEXER_SHUTDOWN_DRAIN_TIMEOUT, 300s default) so shutdown stays within the deployment's grace period.
• Indexer: an empty batch with no open workbench is dropped instead of creating one just to commit it empty.

Tests

• test_source_actor_drain: a source without ack state exits immediately on drain, after emitting the flush batch.
• Full indexing_service + indexing_pipeline suites green (non-opt-in sources keep plain kill semantics).
• End-to-end drain coverage lands with the first ack-based source (stacked PR feat(indexers): add a scalable NATS JetStream source #4).

🤖 Generated with Claude Code

https://claude.ai/code/session_01QHjEoDvmLtHNpEUbkUY56D