[Data] Prevent output backpressure from starving lineage reconstruction

[dragongu]

Description

On preemptible / elastic clusters (spot workers preempted often), Ray Data
pipelines that rely on lineage reconstruction can collapse to near-zero
goodput after a recoverable object loss. The streaming generator that
reconstruction is waiting on is starved by its own operator's output
backpressure — which is in turn caused by the very downstream task that is
blocked on that reconstruction.

Consider a typical elastic-resource pipeline (worker pods can be preempted at
any time):

read_parquet ──► map_batches A ──► map_batches B ──► write_parquet
                 (streaming gen)     (actor)

  A_task_k  yields intermediate block A1
                     │
                     ▼
  B_task    consumes A1, produces B1
                     │
              node preempted
                     ▼
              B1 is LOST  ──►  lineage reconstruction of B1 needs A1,
                               and A1 was produced by A_task_k

The starvation cycle

A_task_k's original attempt is still running, so Ray Core queues it for
resubmit (replay) and waits for the current attempt to return. For that attempt
to return, Ray Data must keep reading A_task_k's streaming output. But Ray
Data applies output read budgets per operator, and under backpressure the
budget for operator A is 0:

        ┌─────────────────────────────────────────────────────────────┐
        │                                                             │
        ▼                                                             │
  downstream write task is PENDING on B1                              │
  (waiting for B1 to be reconstructed)                                │
        │                                                             │
        │  can't consume from B → B backs up → A backs up             │
        ▼                                                             │
  output backpressure clamps A's read budget → 0                      │
        │                                                             │
        ▼                                                             │
  A_task_k (needed for replay) is starved of reads                    │
        │                                                             │
        ▼                                                             │
  A_task_k's current attempt cannot finish                            │
        │                                                             │
        ▼                                                             │
  replay stays queued → A1 and B1 not reconstructed  ─────────────────┘

This is not a permanent deadlock in the general case: Ray Data's existing
output-backpressure escape hatch (OutputBackpressureGuard.should_unblock)
periodically releases a single block. But its forward progress is only nominal —
it fires on a ~10s idle-detection interval, and it releases the budget to any
ready task, not specifically the generator replay is queued behind. How badly
this bites depends on the resource regime:

• Elastic CPU (spare capacity downstream): the drip does eventually
  reconstruct, but ~1 block / 10s loses the race against sustained preemption —
  goodput collapses.
• Constrained accelerators (e.g. a lowest-priority job whose GPUs are
  preempted, with none free to reschedule on): the released block is often not
  the reconstruction-critical generator, and even when it is, that block has
  nowhere to go — surviving GPU actors are starved waiting for inputs and no
  new actors can be scheduled. The released blocks instead pile up in the object
  store, which tightens the very backpressure starving the generator. Here the
  "one block per round" safety net produces no real forward progress and the
  pipeline effectively hangs — the same cycle, surfacing as a full halt.

Root cause

Ray Data computes output read budgets at the operator level and treats all
of an operator's ready tasks identically. It has no way to tell that one
specific streaming generator is the one Ray Core is waiting on for
reconstruction, so under backpressure that generator is starved along with
everything else. Ray Core cannot resolve this alone either — it does not observe
or control Ray Data's per-operator output-read policy.

Fix

Give Ray Data a minimal, bounded feedback path from Ray Core's reconstruction
state. All of it lands in OutputBackpressureGuard, which already owns the one
existing exception to output backpressure:

1. Expose the signal from Core. Add
   CoreWorker::GetLocalQueuedGeneratorResubmitTaskIds() (aggregating new
   GetQueuedGeneratorResubmitTaskIds() accessors on the normal and actor task
   submitters), surfaced to Python via
   CoreWorker.get_local_queued_generator_resubmit_task_ids(). It returns the
   locally-submitted streaming generators reconstruction has queued for resubmit.
   Local (no RPC) but takes the submitter lock, so it is only queried when at
   least one operator has a finite output read budget.

2. Prioritize the reconstruction-critical generator.
   OutputBackpressureGuard.order_ready_tasks orders an operator's ready tasks
   so queued generators come first, then by task index (deterministic; falls
   back to plain task-index order when nothing is queued).

3. Grant a bounded bypass lane. For a fully-backpressured operator the guard
   vends a ReconstructionBypassLane worth
   DataContext.lineage_reconstruction_backpressure_bypass_blocks blocks (env
   RAY_DATA_LINEAGE_RECONSTRUCTION_BACKPRESSURE_BYPASS_BLOCKS, default 1, 0
   disables), shared across that operator's queued generators. The lane reads one
   block at a time, stops early once the generator has nothing ready, and
   never touches the operator's byte budget — so ordinary tasks on the same
   operator stay fully backpressured.

This breaks the cycle: A_task_k is drained a bounded number of blocks per
round, its current attempt can finish, replay proceeds, A1 and then B1 are
reconstructed, and recovery keeps pace with the preemption rate.

Keeping this in the guard (rather than the scheduling loop) puts both
backpressure exceptions in one place. The lane stays a separate mechanism
from should_unblock because eligibility is task-scoped: widening the
operator-level byte budget would let every task on the operator read past
backpressure, not just the one generator reconstruction is blocked on.

[gemini-code-assist]

Code Review

This pull request introduces a mechanism to prioritize and drain streaming generator tasks queued for resubmission by lineage reconstruction under backpressure. It exposes the queued resubmit task IDs from the core worker task submitters up to the Python streaming executor, where these tasks are prioritized and granted a one-block lane to prevent starvation and deadlock. Feedback suggests adding defensive guards in streaming_executor_state.py to prevent potential AttributeError exceptions when checking the first ready task's ID and type.

[ayushk7102]

Thanks for your PR!

Main feedback: should we fold this as part of the existing BackpressureGuard?

Also, should we should characterize this as a fix addressing stall/performance drop than deadlocking, because we are still making progress

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 2 potential issues.

^{Reviewed by Cursor Bugbot for commit fd8c09e. Configure here.}