[Yin-1022]

Description

DataOpTask currently starts output-backpressure accounting whenever max_bytes_to_read == 0.

However, a zero output-read budget does not always mean that the task still has unread output.
After the stream is exhausted and all block/metadata refs have been consumed, the task may only be waiting for terminal completion. In that state, the terminal-only wait should not contribute to task_output_backpressure_s.

This became more visible while reviewing #67180.
That change allows exhausted tasks to finalize even when the output-read budget is zero, but the existing backpressure accounting still starts before the terminal check.
As a result, the interval between stream exhaustion and terminal completion can still be counted as output backpressure.

In normal execution this interval is usually small, so this is primarily a metrics accounting correctness cleanup.

Changes

• Treat a zero output-read budget as output backpressure only when the task still has unread output.
• Consider output unread when:
  • a block ref is pending,
  • a metadata ref is pending, or
  • the streaming generator has not yet been exhausted.
• Close an existing output-backpressure interval once no unread output remains, even if the read budget is still zero.
• Add regression test coverage for:
  • an exhausted empty task waiting only for terminal completion;
  • a normal unread output under a zero read budget;
  • a pending block/metadata pair after the generator stream is already exhausted.

Before / After Comparison

Before:

stream exhausted
        ↓
no pending block / metadata output
        ↓
max_bytes_to_read = 0
        ↓
output-backpressure timer starts
        ↓
wait for terminal completion
        ↓
terminal-only wait is included in task_output_backpressure_s

After:

max_bytes_to_read = 0
        +
task still has unread / pending output
        ↓
count as output backpressure

If the stream is already exhausted and no block or metadata refs remain pending:

max_bytes_to_read = 0
        +
no unread output
        ↓
terminal-only wait is not counted

Actual unread output continues to be accounted for, including the case where the generator stream is exhausted but a block/metadata pair has already been pulled and remains pending inside DataOpTask.

Related issues

Fixes #67312
Related to #67180.

#67180 fixes task finalization when the output-read budget is zero.
This change is independent and only adjusts the output-backpressure metric so that terminal-only waiting time is not included once no unread output remains.

Additional information

The regression behavior was verified before and after the change:

• terminal-only wait:
  • before the fix: a synthetic 10-second terminal wait was reported as 10.0 seconds of output backpressure;
  • after the fix: the same terminal-only wait reports 0.
• unread output under a zero budget continues to report the expected backpressure duration.
• an exhausted stream with a pending block/metadata pair also continues to report the expected backpressure duration.

The full TestDataOpTask test class passes with the change.

The production change is intentionally small and does not modify task scheduling, output-read budgeting, or terminal-finalization behavior.

[gemini-code-assist]

Code Review

This pull request refactors the end-of-stream terminal checks and output backpressure tracking in DataOpTask to use non-blocking, rate-limited checks and prevent terminal checks from being misclassified as backpressure. It also adds corresponding unit tests. The review feedback suggests initializing _next_terminal_check_s to None instead of 0.0 and updating the rate-limiting check to handle this, preventing potential issues on platforms where time.monotonic() might return negative values.

[Jadiouo]

Verified against base ce4256e and head 7326d2d (Python 3.13, CPU only).

• The repro from #67312 gives _total_output_backpressure_s = 10.0 on base and 0 with the PR.
• The new tests: test_terminal_wait_not_counted_as_output_backpressure fails on base (assert 10.0 == 0), the other two pass on both, and all three pass with the PR. The full test_streaming_executor.py passes (57 passed).
• With a patched clock: a pending block/meta pair after the stream is exhausted still counts, an interval closed by a positive budget is counted as before, and I saw no double counting.

The gemini comments about _next_terminal_check_s refer to the earlier commit 9b2e18f and don't apply to the current diff.

LGTM.

[owenowenisme]

LGTM