Keep jobs running across queue restarts

[taagarwa-rh]

Primarily written with GPT 5.6 Sol and OpenCode

[coderabbitai]

Important

• 🔍 Trigger review

This repository does not receive automatic reviews because it has fewer than 10 stars.

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Team

Run ID: 78b43a03-4d23-43ec-bb12-7e4ff6017e47

---

_{Comment @coderabbitai help to get the list of available commands.}

[rounakbende10]

Code Review: Keep jobs running across queue restarts

Reviewed: all 4 changed files (460+, 127-). Overall this is a well-structured approach to job recovery -- the new COMPLETING/FAILING transitional states, deterministic instance naming, and retry-until-success cleanup loops are sound design choices. Below are the issues I found, ordered by severity.

---

Critical (confidence 90-100)

1. adopt_running_instance is not protected by self._lock (confidence: 95)

acquire_instance correctly holds self._lock for the entire provision-and-track sequence, but adopt_running_instance mutates self._instances without acquiring the lock. If the idle_cleanup_loop runs concurrently (it holds the lock while iterating _instances), this creates a race: the cleanup loop could delete or evict the instance between the instance_exists check and the dict write, or the dict mutation could interfere with the cleanup iteration.

# Current (no lock):
async def adopt_running_instance(self, model_name, gpu_config):
    instance_name = self._pick_instance_name()
    if not await self._manager.instance_exists(instance_name):
        raise ...
    self._instances[instance_name] = NebiusInstanceState(...)
    return instance_name

Fix: Wrap the body in async with self._lock:, matching acquire_instance.

---

2. Double _active_job assignment creates a stale-reference window (confidence: 92)

In _worker, _active_job is set before _process_queued_job is awaited, and then _process_queued_job also sets _active_job at its start. The OpenshiftJob object created in _worker (line ~603 in the diff) is a throwaway -- it is never used for actual oc commands. But between the moment _worker sets _active_job and _process_queued_job overwrites it, if a DELETE /jobs/{job_id} request arrives, the cancellation handler will call _active_job[1].cancel() on the outer task, which is the processing task -- this part works. However the _active_job[2] (the OpenshiftJob) reference is the stale throwaway, so any code that uses _active_job[2] for cleanup before _process_queued_job has a chance to overwrite it would operate on the wrong object.

This is not immediately exploitable because delete_job only uses _active_job[0] (job_id) and _active_job[1] (task), but it is a latent bug if future code trusts _active_job[2].

Fix: Remove the _active_job assignment from _worker and let _process_queued_job be the sole owner. Or remove _active_job from _worker entirely and only set it in _process_queued_job.

---

Important (confidence 80-89)

3. _retry_cancellation and _retry_terminal_job can spin forever if cleanup is permanently broken (confidence: 88)

Both retry loops run indefinitely with a 5-second sleep when cleanup keeps failing. If the OpenShift cluster is misconfigured or the namespace is deleted, these loops block the serial job queue forever -- no subsequent job can ever start.

async def _retry_terminal_job(...):
    while not await _finish_terminal_job(...):
        await asyncio.sleep(5)

Suggestion: Add a maximum retry count or a total timeout (e.g., 10 minutes). After exhausting retries, force the terminal status with a cleanup-failed error and move on. This preserves the serial queue's liveness.

---

4. _delete_recovered_nebius also spins forever (confidence: 87)

Same pattern as above:

async def _delete_recovered_nebius(job_id):
    while True:
        try:
            await _nebius.delete_recovered_instance()
            return
        except Exception:
            await asyncio.sleep(5)

If the Nebius API is permanently unreachable, this blocks the startup path or the processing of recovered terminal jobs indefinitely.

Suggestion: Same as above -- add a retry cap. At worst, log a loud warning and let the idle cleanup loop handle it later.

---

5. Removal of shutdown Nebius cleanup is intentional but undocumented (confidence: 85)

The old code deleted all Nebius instances on graceful shutdown. The new code removes this entirely, relying on the idle cleanup loop to eventually delete VMs. This is the correct trade-off for surviving restarts (you don't want to kill VMs that active jobs are using), but it means that if the queue service is permanently decommissioned (not just restarted), Nebius VMs will leak until the idle timeout fires. Since terminationGracePeriodSeconds: 60 and Recreate strategy are set, there is a window during rolling restarts where no cleanup loop is running.

Suggestion: Add a comment explaining this design choice. Consider a separate "drain" endpoint or an explicit "delete all VMs" admin action for decommissioning.

---

6. _delete_harbor_pods iterates all harbor pods in the namespace (confidence: 82)

The new implementation fetches all pods matching app=harbor,harbor-session and then filters by HARBOR_PARENT env var in Python. For namespaces with many concurrent jobs (or leftover pods), this could be slow and memory-intensive compared to the old label-selector approach. The clean_legacy_pods fallback (cleaning pods that lack HARBOR_PARENT entirely) is pragmatic for migration but could accidentally delete pods from other jobs that haven't been updated yet.

) or (self._clean_legacy_pods and not has_parent):
    pod_names.append(pod["metadata"]["name"])

Suggestion: After migration stabilizes, remove the clean_legacy_pods fallback or gate it behind a flag to prevent cross-job interference.

---

7. is_resume variable used before definition in _process_queued_job for non-Nebius non-adopt path (confidence: 80)

In _process_queued_job, is_resume is only defined inside the elif nebius_gpu_config is not None and _nebius: block (the non-adopt Nebius provisioning branch). But at line ~526, _run_job is called with openrouter=is_openrouter(server_url) -- this is fine. However, the flow through the function for a non-Nebius, non-adopt job skips all the blocks that define model_config, so model_config remains None and the --model-max-len check is safely skipped. This is correct but fragile -- is_resume is scoped to a branch that only executes for Nebius jobs, making the code hard to follow.

---

Minor observations (not blocking)

• The exec env change in the deployment YAML is good -- ensures SIGTERM propagates to uvicorn rather than the parent shell.
• backoffLimit: 0 on Job specs is a good addition -- prevents Kubernetes from restarting failed pods, which would confuse the monitoring loop.
• The _build_parent_env_shell_step helper for resume jobs correctly propagates HARBOR_PARENT into the resumed config.
• The CancelledError handling in _run_oc_command (job.py) correctly terminates the subprocess before re-raising -- good defensive coding.
• ORDER BY rowid on list queries is a nice touch for deterministic ordering.

Summary

The core recovery logic is sound: jobs persist their state in SQLite, the new process re-enqueues non-terminal jobs, and the COMPLETING/FAILING states prevent double-cleanup. The main concerns are (1) the missing lock on adopt_running_instance, (2) infinite retry loops that can block the queue, and (3) the double _active_job assignment pattern. Items 1 and 2 should be fixed before merge; item 3 is a latent risk worth cleaning up.

[taagarwa-rh]

Implemented in separate commits:

• f0b98cf Lock Nebius instance adoption.
• 779aaa4 Remove stale _active_job references.
• df263e4 Bound cleanup retries with configurable limits.
• 7d6547c Document shutdown cleanup policy.
• 88c6368 Scope legacy Harbor cleanup by parent label.
• c30d2ea Isolate recovery test imports.
• b0c6e05 Fix discovered OpenRouter environment overwrite.

Verification:

• Claims 1–3: confirmed.
• Claim 4: confirmed for recovered jobs; startup deletion itself runs in the background.
• Claim 5: confirmed, except permanent decommissioning leaks indefinitely rather than until idle timeout.
• Claim 6: scan is necessary for environment-based ownership; legacy cross-job risk fixed.
• Claim 7: not a bug; is_resume is only used inside its defining branch.

[rounakbende10]

1. Unbounded retry loops in adoption paths

The two while True loops around _get_job() in _run_job and _process_queued_job have no max attempt bound. If the OpenShift API is unreachable (without triggering a pod restart), the serial worker blocks indefinitely — no other jobs can proceed.

CancelledError does propagate through except Exception (it's a BaseException in Python 3.9+), so shutdown/user-cancel will break out. But sustained API outage during recovery stalls all jobs silently.

The cleanup paths in this same PR correctly use CLEANUP_MAX_ATTEMPTS — these loops should too for consistency and safety.

2. CancelledError swallowed in _run_job — Nebius VMs leak ~10 min GPU time

When a user cancels a running Nebius job, _run_job catches CancelledError, calls _finish_cancellation + _retry_cancellation, but does not re-raise. Control returns to _process_queued_job, which skips its own CancelledError handler (where _delete_recovered_nebius would run) and falls through to mark_job_completed — marking the VM as idle instead of deleting it immediately.

The VM then lingers until NEBIUS_IDLE_TIMEOUT (600s), wasting ~10 min of GPU time per cancelled job.

Suggested fix: re-raise CancelledError after the cancellation cleanup in _run_job, or add a status check before mark_job_completed to skip it for cancelled jobs.

[taagarwa-rh]

Both claims were verified and fixed in separate commits:

• c662dd8 Bound recovered job inspection retries
• 98ed20f Delete Nebius instance on job cancellation