enhancement(nats source): call drain() during shutdown

[benjamin-awd]

Summary

This PR enables the NATS source to drain messages when shutting down. Currently we unsubscribe, which is not as robust as draining before disconnect since inflight messages can be dropped.

Closing a connection (using close()), or unsubscribing from a subscription, are generally considered immediate requests. When you close or unsubscribe the library will halt messages in any pending queue or cache for subscribers. When you drain a subscription or connection, it will process any inflight and cached/pending messages before closing.

More details here: https://docs.nats.io/using-nats/developer/receiving/drain

How did you test this PR?

Ran this on a relatively high velocity NATS subject (messages redacted, but timestamp is the native timestamp created by Vector) with a console sink to stdout.

With this change, events are still handled even after the shutdown is triggered.

{"timestamp":"2025-08-22T11:00:58.831517Z", "payload: foo"}
{"timestamp":"2025-08-22T11:00:58.834378Z", "payload: bar"}
^C
2025-08-22T11:00:58.860597Z  INFO vector::signal: Signal received. signal="SIGINT"
2025-08-22T11:00:58.860687Z  INFO vector: Vector has stopped.
2025-08-22T11:00:58.860808Z  INFO source{component_kind="source" component_id=private component_type=nats}: vector::sources::nats::source: Shutdown signal received. Draining NATS subscription...
2025-08-22T11:00:58.862056Z  INFO vector::topology::running: Shutting down... Waiting on running components. remaining_components="print, process_message, private, nats-private-v1" time_remaining="59 seconds left"
2025-08-22T11:00:58.876921Z  INFO source{component_kind="source" component_id=private component_type=nats}: vector::sources::nats::source: NATS source drained and shut down gracefully.
2025-08-22T11:00:58.877080Z  INFO async_nats: event: closed
{"timestamp":"2025-08-22T11:00:58.876851Z", "payload: baz"}
2025-08-22T11:00:58.877768Z  INFO async_nats: event: closed

Change Type

• Bug fix
• New feature
• Non-functional (chore, refactoring, docs)
• Performance

Is this a breaking change?

• Yes
• No

Does this PR include user facing changes?

Not sure about this one, I guess technically it's user-facing?

• Yes. Please add a changelog fragment based on our guidelines.
• No. A maintainer will apply the no-changelog label to this PR.

References

• Related to Does the NATS source drain messages when shutdown signal is given? #21756

[pront]

Nats integration tests failed: https://github.com/vectordotdev/vector/actions/runs/17244108073/job/48929272324

[benjamin-awd]

I think the new test had some kind of race condition, made a small tweak which will hopefully deal with that