[Jobs] Guard recovered monitors against stale job state

[zzchun]

Summary

Prevent recovered submission-job monitors from recreating deleted jobs or
overwriting jobs that have already reached a terminal state.

Problem

JobManager._recover_running_jobs() schedules monitors asynchronously for
non-terminal jobs. A job record can be deleted or transition to a terminal
state between the recovery scan and the monitor's reconciliation pass.

The existing monitor may then attempt to write FAILED:

• If the record was deleted, put_status() may recreate it with
  entrypoint="Entrypoint not found.".
• If the job is already terminal, the monitor may attempt an invalid or
  incorrect terminal-state transition.
• A temporary get_info() failure for a PENDING job may be treated as a
  monitor failure instead of being retried.

Changes

• Stop monitoring immediately when the job record no longer exists.
• Preserve terminal job states without writing another status.
• Apply the same deleted/terminal checks in the monitor exception path.
• Retry temporary PENDING job metadata-read failures on the next monitor
  iteration.
• Add guarded status updates that require:
  • the job record to still exist; and
  • the current status to match the status observed by the monitor.
• Stop the monitor when a guarded update loses a deletion or status race.
• Add focused tests for deleted records, terminal transitions, temporary GCS
  failures, and guarded

Related issues

Fixes #65251

Additional information

Optional: Add implementation details, API changes, usage examples, screenshots, etc.

[gemini-code-assist]

Code Review

This pull request introduces guarded status updates to the job manager to prevent race conditions when updating job statuses, particularly during recovery or concurrent deletions. It updates put_status to check if the job info exists and matches an expected status before writing, and adds comprehensive unit tests for these scenarios. The reviewer identified a critical issue in _monitor_job_internal where returning instead of continuing when a status update fails (due to a concurrent transition from PENDING to RUNNING) would stop monitoring the job entirely, leaving it unmonitored.

[gemini-code-assist]

If status_updated is False because the job status transitioned from PENDING to RUNNING concurrently, calling return here will stop the monitor and leave the running job completely unmonitored.

Instead of returning, we should continue the loop. In the next iteration, get_status will fetch the updated RUNNING status, and the monitor will correctly proceed to track the supervisor actor.

Suggested change

	if not status_updated:
	logger.info(
	"Stopping monitoring for job %s because its job "
	"info was deleted or its status changed.",
	job_id,
	)
	return
	if not status_updated:
	logger.info(
	"Retrying monitoring for job %s because its job "
	"info was deleted or its status changed.",
	job_id,
	)
	continue

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

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

[cursor]

Stale ping reused after retry

High Severity

After a guarded exception update loses its race, the monitor continues without clearing ping_obj_ref. A failed ping from the prior status can be reused immediately on the next iteration, re-raise, and mark the job FAILED even though it has advanced to a new non-terminal state such as RUNNING.

Additional Locations (1)

• python/ray/dashboard/modules/job/job_manager.py#L336-L345

 

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