[Core] Release the finished attempt's pending count when resubmitting a generator

[LuciferYang]

Description

TaskManager::MarkGeneratorFailedAndResubmit runs when an intermediate object of a still-running streaming generator is lost and needed again for recovery. The entry is SUBMITTED_TO_WORKER at that point, so num_pending_tasks_ already counts it. SetTaskStatus does not touch that counter, and SetupTaskEntryForResubmit then adds one unconditionally, so the task ends up counted twice with a single attempt in flight. Neither CompletePendingTask nor FailPendingTask runs on this path, so nothing gives the count back: the resubmitted attempt's completion decrements once and leaves num_pending_tasks_ permanently high by one, growing by another one on every recovery.

An inflated count makes CoreWorker::IsIdle() return false forever, so the raylet never reclaims an otherwise idle worker. It also blocks DrainAndShutdown and ShutdownIfNeeded, which wait for the count to reach zero, and makes NumPendingTasks() over-report, which surfaces in job info as a job that always has tasks running.

This releases the count before SetupTaskEntryForResubmit adds it back for the new attempt. The increment there is correct for the other caller, ResubmitTask, which reaches it with a FINISHED or FAILED task whose count was already released, so the decrement belongs in the generator path rather than in the helper. Both happen under one mu_ hold, so no reader observes the dip.

It also asserts the entry is still pending, matching RetryTaskIfPossible and FailPendingTask. num_pending_tasks_ is a size_t, and this was the only decrement site without that check.

Related issues

Fixes #65313

Additional information

The test runs two recovery cycles, since the miscount was additive: without the fix the count reaches 2 after the first cycle and 3 after the second. It also asserts the retry callback fired on each cycle, so the count staying at 1 cannot come from a change that skips the resubmit instead of fixing the accounting. I verified the direction by reverting only the decrement and re-running: the test fails with Which is: 2, and the fixture's existing leak check fails too because the count never returns to zero.

bazel test //src/ray/core_worker/tests:task_manager_test passes.

I used AI assistance to investigate and draft this change. I reviewed every changed line and ran the tests myself.

[gemini-code-assist]

Code Review

This pull request fixes a pending task count leak during generator resubmission in TaskManager::MarkGeneratorFailedAndResubmit. It adds a check to ensure the task entry is pending and decrements num_pending_tasks_ to release the finished attempt's count before resubmitting. A unit test has also been added to verify that the pending task count remains correct across recovery cycles. There are no review comments, so we have no feedback to provide.

[github-actions]

This pull request has been automatically marked as stale because it has not had
any activity for 14 days. It will be closed in another 14 days if no further activity occurs.
Thank you for your contributions.

You can always ask for help on our discussion forum or Ray's public slack channel.

If you'd like to keep this open, just leave any comment, and the stale label will be removed.

[LuciferYang]

rebased