Fix streaming token ids data loss under load

[vladnosiv]

When handling multiple scheduler batches before _wait_one_response is naturally scheduled, intermediate outputs accumulate in state.out_list. Previously, the code consumed only the last element (state.out_list[-1]).

While this masked the issue for cumulative text consumers, it introduced a severe bug for consumers reading output_ids limits. Since streaming token IDs are emitted as disjoint deltas, dropping intermediate entries from out_list caused silent, unrecoverable data loss (missing tokens, missing format in tool-calling, and output corruption).

This race condition is significantly amplified under high concurrency, especially when using --skip-tokenizer-init (e.g., in Dynamo environments), where the absence of a CPU-bound detokenizer allows ZMQ messages to arrive at wire speed and accumulate rapidly.

Proposed Changes

• Atomically drain out_list: Differentiate between streaming and non-streaming requests. For streams, we now drain and yield all pending output dicts sequentially. For non-streams, we preserve the previous behavior of only taking the latest cumulative output (state.out_list[-1:]).
• Preserve Finalization Logic: Ensured that request logging, metrics exporting, and abort handling still only trigger on the very last chunk (is_last) when the state is finished.
• Add Observability: Added a warning when draining multiple queued chunks in streaming mode. This provides crucial observability into back-pressure dynamics and helps explain P99 TBT spikes.

Resolves #19976

cc @ishandhanani

[gemini-code-assist]

Warning

You have reached your daily quota limit. Please wait up to 24 hours and I will start processing your requests again!

[ishandhanani]

Thanks for the fix @vladnosiv this is important.

For completeness can you add a comment or update the PR description with a before/after

[vladnosiv]

Thanks for the fix @vladnosiv this is important.

For completeness can you add a comment or update the PR description with a before/after

added a description in the PR, described the problem in more detail in the linked issue

[ishandhanani]

/tag-and-rerun-ci

[ishandhanani]

Only a single CI test failed in the previous run

[vladnosiv]

@hnyls2002 Hi !
Please take a look

[vladnosiv]

Seems like in registered/distributed/test_dp_attention.py test the server didn't even start
On my side test crashes due to using the lmsys/sglang-ci-dsv3-test private model, which I do not have access to
In any case, it looks irrelevant to the current PR, because the problem is still at the start of the test