[fix](pipeline) Prevent finalized tasks from being rescheduled

[mrhhsg]

What problem does this PR solve?

Issue Number: None

Problem Summary: A delayed dependency wake-up can leave a terminal pipeline task in the runnable queue. The stale task may be executed again after its shared operator state has been released, causing invalid virtual calls or use-after-free.

This change:

• keeps queued tasks and their fragment contexts alive until the queue entry is consumed;
• stores dependency waiters weakly and pins them only while delivering a wake-up;
• atomically claims PipelineX tasks and drops terminal tasks before accessing execution state;
• drains pending queue entries outside the queue lock during shutdown.

Release note

None

Check List (For Author)

• Test:
  • Unit Test: ./run-be-ut.sh --run --filter='TaskQueueTest.*:TaskSchedulerTest.*' -j 24
  • Regression test: Not required for this scheduler lifecycle fix
• Behavior changed: No
• Does this need documentation: No

[hello-stephen]

Thank you for your contribution to Apache Doris.
Don't know what should be done next? See How to process your PR.

Please clearly describe your PR:

1. What problem was fixed (it's best to include specific error reporting information). How it was fixed.
2. Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
3. What features were added. Why was this function added?
4. Which code was refactored and why was this part of the code refactored?
5. Which functions were optimized and what is the difference before and after the optimization?

[mrhhsg]

/review

[github-actions]

Automated review result: REQUEST_CHANGES

Critical checkpoints:

• Correctness and lifecycle: blocked. A losing duplicate wake token can cross into a later blocked or PENDING_FINISH epoch, causing premature close or duplicate dependency registration.
• Concurrency and scheduler accounting: blocked. The claim happens after dequeue mutates plain task fields, and immediate requeue can hot-spin while racing the active worker's statistics update.
• Ownership and shutdown: the fragment-context alias holder, weak dependency waiters, queue-size balance, and out-of-lock close drain are sound on the traced production paths.
• Compatibility and tests: all discovered TaskQueue implementations/callers use the new holder type and CMake discovers the test, but the tests do not exercise live duplicates, dependency blocks, PENDING_FINISH, stealing, or concurrent close.
• User focus: no additional user-provided review focus was supplied; the whole PR was reviewed.

Review completion: all nine changed files and related control flow were reviewed. Two normal full-review agents plus a separate risk-focused agent converged in Round 2 with NO_NEW_VALUABLE_FINDINGS against this exact head and comment set. Static review only; builds/tests were not run as required. No repository code-review SKILL.md was present after exhaustive in-checkout search, so the authoritative review prompt was followed directly.

[github-actions]

[P1] Do not recycle a wake token across task-state epochs

Requeuing the losing entry preserves it without tying it to the dependency epoch that created it. A delayed duplicate can sit in the queue while the owner runs; if the owner then reaches EOS, is_pending_finish() registers an unready finish dependency and the owner publishes PENDING_FINISH before clearing _running. This stale entry can now claim the task, and the PENDING_FINISH branch calls _close_task() without rechecking finish dependencies for PipelineX, releasing async-writer/exchange state before the callback makes the dependency ready. If the owner instead enters an ordinary blocked state, the stale entry re-enters the same unready dependency and trips _add_block_task's duplicate DCHECK (or appends duplicate waiters in release builds). Please coalesce or generation-tag pending wakes so an old token cannot satisfy a later wait epoch, and add a latched regression covering both transitions.

[github-actions]

[P2] Claim or coalesce duplicates before touching task accounting

The atomic claim happens only after take() has updated this task's _queue_level/_core_id and stopped its wait-worker watcher. When this branch loses, it immediately pushes the same token again, so an idle worker can hot-loop pop/push for the owner's full execution slice (or longer close) while push() reads _runtime, increments _schedule_time, and restarts the same plain stopwatch. Meanwhile the owner calls update_statistics(), which writes _runtime and reads _core_id/_queue_level; per-core queue locks do not synchronize these task fields, especially after steals. This leaves C++ data races plus corrupted priority/profile accounting even though execute() is serialized. Please claim/coalesce before per-task dequeue accounting and add a latched multi-worker duplicate test.