fix(jobs): reap RUNNING jobs that never stored an event - #318
Conversation
The ghost-job reaper's _find_stale_jobs query was driven from job_events with an INNER JOIN job_info, so a job that entered RUNNING but never persisted an event produced zero rows and was invisible to the reaper. That is exactly the failure it exists to catch: the runner marks a job RUNNING before the first event is stored (and BatchingEventStore batches writes), so a worker crash/OOM in that window left the job stuck in RUNNING forever — status polling never reached a terminal state and the job permanently counted against the active-job cap. Drive the query from job_info LEFT JOIN job_events and use COALESCE(MAX(je.created_at), ji.updated_at) as the staleness clock, so zero-event RUNNING jobs fall back to job_info.updated_at (set when the job entered RUNNING). Jobs that did emit events keep the existing last-event-age behavior. Also guard for a missing job_info table. Verified the SQLite/Postgres timestamp columns compare correctly. Adds a real-DB regression suite (frontends/aiq_api/tests/test_ghost_reaper.py) covering the zero-event ghost, the within-timeout fresh job, stale/recent events, non-running jobs, missing tables, and a mixed fleet. The three zero-event assertions fail against the old INNER JOIN and pass with the fix. Full frontends/aiq_api suite passes; ruff clean. Closes NVIDIA-AI-Blueprints#317 Signed-off-by: Torkian <torkian@mac.com>
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughThe ghost-job reaper now detects zero-event RUNNING jobs using ChangesGhost Job Reaper and Running Lease
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Worker as run_agent_job
participant JobInfo as job_info
participant Reaper as ghost reaper
participant JobEvents as job_events
Worker->>JobInfo: set status to RUNNING
Worker->>JobInfo: refresh updated_at periodically
Reaper->>JobInfo: find stale RUNNING jobs
Reaper->>JobEvents: inspect latest event timestamps
Reaper->>JobInfo: conditionally set stale job to FAILURE
Reaper->>JobEvents: record GhostJobTimeout error
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@frontends/aiq_api/src/aiq_api/routes/jobs.py`:
- Around line 1187-1194: The job reaper guard in jobs.py is checking for
job_events after EventStore._ensure_table_exists(db_url) has already created it,
so that part of the condition is redundant. Update the logic around the
EventStore._ensure_table_exists, EventStore._get_or_create_sync_engine, and
inspector.has_table checks so only the meaningful job_info existence check
remains, or move the table-creation call after the guard if you want the comment
to stay accurate. Make sure the final condition matches the intended behavior
described in the surrounding comment.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Enterprise
Run ID: 3ce28538-65c9-4425-82eb-edfa4e9fe6d3
📒 Files selected for processing (2)
frontends/aiq_api/src/aiq_api/routes/jobs.pyfrontends/aiq_api/tests/test_ghost_reaper.py
📜 Review details
🧰 Additional context used
📓 Path-based instructions (4)
**/*.py
📄 CodeRabbit inference engine (CONTRIBUTING.md)
Run ruff check and ruff format validation for Python code changes
**/*.py: Python code must be linted and formatted with Ruff using line length 120, target Python 3.11, rule sets E,F,W,I,PL,UP, and isort force-single-line configuration
Never commit secrets, tokens, or environment-specific hostnames; use environment variables and SecretStr instead, resolving API keys at runtime
Never print or log secret values, including in tool output or error messages
Missing-secret paths must degrade gracefully (stub/skip), not crash or leak
Do not hand-reformat unrelated code when making changes; match the existing import and formatting style
Files:
frontends/aiq_api/tests/test_ghost_reaper.pyfrontends/aiq_api/src/aiq_api/routes/jobs.py
**/*test*.py
📄 CodeRabbit inference engine (CONTRIBUTING.md)
Run pytest for all behavior changes in Python code
Files:
frontends/aiq_api/tests/test_ghost_reaper.py
**
⚙️ CodeRabbit configuration file
**:AI-Q Agent Guidance
Repository-global instructions for coding agents and for humans reviewing
agent-authored changes. These rules apply to every task in this repository.
Task-specific runbooks live in.agents/skills/— load the
relevant skill before starting a workflow it covers.Project overview
AI-Q is an NVIDIA AI Blueprint: an enterprise research agent built on the
NeMo Agent Toolkit (NAT). The deployed product is a research blueprint, not
a general skill runtime. New retrieval sources and tools are NAT functions;
agent behavior is driven by workflow YAML, Jinja2 prompts, and a data-source
registry — not by hard-coded logic.Primary boundaries:
- Backend Python package:
src/aiq_agent/.- Data-source and tool packages:
sources/(each is its own package).- Frontends and tooling:
frontends/(web UI infrontends/ui/, eval harnesses
infrontends/benchmarks/).- Configs, deployment, docs:
configs/,deploy/,docs/.Stay inside this repository. If your workspace also contains adjacent repos
(for example a sibling NeMo-Relay checkout), do not edit them as part of an AI-Q
change. Treatsources/*as independent packages: prefer the smallest change
scoped to the package you are touching.Repository structure
Path Purpose src/aiq_agent/Backend agent, FastAPI extensions, auth, observability, knowledge sources/Data-source / tool packages (e.g. tavily_web_search,google_scholar_paper_search)configs/Workflow YAML configs (e.g. config_cli_default.yml)frontends/ui/Next.js / React / TypeScript / Tailwind / KUI web UI frontends/benchmarks/Eval harnesses: freshqa,deepsearch_qa,deepresearch_benchdeploy/Docker Compose and Helm/Kubernetes assets; deploy/.envfor secretsdocs/source/...
Files:
frontends/aiq_api/tests/test_ghost_reaper.pyfrontends/aiq_api/src/aiq_api/routes/jobs.py
{src/aiq_agent/fastapi_extensions/**,frontends/aiq_api/src/aiq_api/**}
⚙️ CodeRabbit configuration file
{src/aiq_agent/fastapi_extensions/**,frontends/aiq_api/src/aiq_api/**}: Treat API, auth, and job-runner changes as externally visible contracts. Check authorization boundaries,
request tracing, async job lifecycle, websocket reconnect behavior, error responses, and cross-user data isolation.
Require tests for route behavior, access decisions, and job state transitions when those surfaces change.
Files:
frontends/aiq_api/src/aiq_api/routes/jobs.py
🧠 Learnings (1)
📚 Learning: 2026-06-14T17:49:00.640Z
Learnt from: torkian
Repo: NVIDIA-AI-Blueprints/aiq PR: 273
File: frontends/aiq_api/tests/test_sse_reconnect_cursor.py:384-401
Timestamp: 2026-06-14T17:49:00.640Z
Learning: When using `unittest.mock.patch` for code that imports dependencies inside functions/generators (e.g., inside `aiq_api.routes.jobs`), don’t patch via an attribute that doesn’t exist on the consuming module. If the generator does `from ..jobs.event_store import EventStore` inside the generator body, then `aiq_api.routes.jobs` will not have an `EventStore` attribute; patch the source class/method in its defining module instead (e.g., `aiq_api.jobs.event_store.EventStore.get_events_async`). Patching `aiq_api.routes.jobs.EventStore...` would raise `AttributeError` because that symbol is not present at module scope.
Applied to files:
frontends/aiq_api/tests/test_ghost_reaper.py
🪛 ast-grep (0.44.1)
frontends/aiq_api/tests/test_ghost_reaper.py
[warning] 46-46: The function mktemp is deprecated. When using this function, it is possible for an attacker to modify the created file before the filename is returned. Use NamedTemporaryFile() instead and pass it the delete=False parameter.
Context: tempfile.mktemp(suffix=".db")
Note: [CWE-377]: Insecure Temporary File [OWASP A01:2021]: Broken Access Control
(avoid-mktemp-python)
[warning] 130-130: The function mktemp is deprecated. When using this function, it is possible for an attacker to modify the created file before the filename is returned. Use NamedTemporaryFile() instead and pass it the delete=False parameter.
Context: tempfile.mktemp(suffix=".db")
Note: [CWE-377]: Insecure Temporary File [OWASP A01:2021]: Broken Access Control
(avoid-mktemp-python)
[info] 46-46: Make sure temporary files are secure
Context: tempfile.mktemp(suffix=".db")
Note: [CWE-377] Insecure Temporary File.
(mktemp)
[info] 130-130: Make sure temporary files are secure
Context: tempfile.mktemp(suffix=".db")
Note: [CWE-377] Insecure Temporary File.
(mktemp)
🔇 Additional comments (3)
frontends/aiq_api/src/aiq_api/routes/jobs.py (2)
1236-1240: LGTM!
1205-1228: 🗄️ Data Integrity & IntegrationAdd PostgreSQL coverage for the stale-job reaper
frontends/aiq_api/src/aiq_api/routes/jobs.py:1205-1228is only exercised by SQLite tests. Add a PostgreSQL case, or document that this path is intentionally SQLite-only, so theNOW()/datetime('now', ...)split and timestamp comparison semantics are covered.frontends/aiq_api/tests/test_ghost_reaper.py (1)
89-146: LGTM!
_ensure_table_exists already creates job_events before the guard, so the
has_table("job_events") disjunct was always false. Keep only the
meaningful job_info check; the comment now matches the code.
Signed-off-by: Torkian <torkian@mac.com>
|
/ok to test 6fbf847 |
…tives Addresses review feedback on the ghost reaper: it could classify a live worker as a ghost during a long cold start. The runner marks a job RUNNING before config/provider/tool/telemetry/MCP/sandbox initialization and before it stores any event or starts heartbeats, so a slow-but-live worker had no signal and could be marked FAILURE while still working. Two complementary fixes: - Lease: from the moment a job enters RUNNING, the worker refreshes a lightweight lease (job_info.updated_at, the column the reaper already falls back to for a zero-event job) on a fixed interval, until the job ends. A live worker in a long init keeps its lease fresh and is not detected as stale; a genuinely dead worker's lease goes stale and it is reaped as before. The refresh is scoped to status='running' so it can never resurrect a terminal job's timestamp. - Atomic transition: the reaper now flips RUNNING -> FAILURE with a single conditional UPDATE (WHERE status='running'), so a job that reached a terminal state between detection and reaping (e.g. a slow worker that finished) is never clobbered; the error event is only emitted when this call actually performed the transition. Adds regression tests: a long-created job with a fresh lease is not reaped, a stale-lease zero-event job still is, the conditional transition flips a running job but leaves a SUCCESS/absent job untouched, and the runner lease touch refreshes only running jobs. Full frontends/aiq_api suite passes (317); ruff clean. Signed-off-by: Torkian <torkian@mac.com>
…// urls Review-pass hardening on the cold-start fix: - Run the lease refresher on a dedicated thread instead of an asyncio task. Cold-start init has synchronous chunks (config load, agent import) that can hold the worker event loop; an on-loop lease could be starved by exactly the slow init it is meant to cover. A thread is immune to that. - Guard the worker's terminal SUCCESS write: if the reaper already marked the job FAILURE (e.g. it was slow to finish), don't resurrect it with SUCCESS. Terminal state is now first-writer-wins on both sides — the reaper's RUNNING->FAILURE is an atomic conditional UPDATE, and the worker skips its success write once the job is terminal. - Accept the legacy postgres:// scheme (not just postgresql://) when choosing NOW() vs CURRENT_TIMESTAMP, so the reaper query and lease work on either connection string. Adds tests: the lease thread actually refreshes a running job's lease and exits promptly on stop, and the dialect helper maps both postgres schemes. Full frontends/aiq_api suite passes (319); ruff clean. Signed-off-by: Torkian <torkian@mac.com>
|
@AjayThorve pushed the fix for the cold-start false-positive (thread-based lease + atomic conditional transition + terminal-state guard on the worker side), with regression tests. Full backend suite green, ruff clean — ready for another look whenever you have a moment. |
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@frontends/aiq_api/src/aiq_api/jobs/runner.py`:
- Around line 898-913: Replace the read-then-write SUCCESS flow in the job
completion path with an atomic compare-and-set operation that updates status and
output only when the current status is RUNNING. Remove reliance on the separate
current_job check, ensure update_job_output propagates the status predicate
through the database write, and add a regression test covering a reaper FAILURE
interleaving so SUCCESS cannot resurrect the job.
In `@frontends/aiq_api/tests/test_ghost_reaper.py`:
- Line 253: Update the test’s runner.LEASE_REFRESH_INTERVAL_SECONDS assignment
to use a small positive interval instead of zero, preserving lease-refresh
verification while preventing a busy loop and SQLite churn during the test.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Enterprise
Run ID: a65416c6-d3aa-464a-af20-83f520fa949d
📒 Files selected for processing (3)
frontends/aiq_api/src/aiq_api/jobs/runner.pyfrontends/aiq_api/src/aiq_api/routes/jobs.pyfrontends/aiq_api/tests/test_ghost_reaper.py
📜 Review details
🧰 Additional context used
📓 Path-based instructions (3)
**/*.py
📄 CodeRabbit inference engine (CONTRIBUTING.md)
Run ruff check and ruff format validation for Python code changes
**/*.py: Python code must be linted and formatted with Ruff using line length 120, target Python 3.11, rule sets E,F,W,I,PL,UP, and isort force-single-line configuration
Never commit secrets, tokens, or environment-specific hostnames; use environment variables and SecretStr instead, resolving API keys at runtime
Never print or log secret values, including in tool output or error messages
Missing-secret paths must degrade gracefully (stub/skip), not crash or leak
Do not hand-reformat unrelated code when making changes; match the existing import and formatting style
Files:
frontends/aiq_api/src/aiq_api/jobs/runner.pyfrontends/aiq_api/tests/test_ghost_reaper.pyfrontends/aiq_api/src/aiq_api/routes/jobs.py
{src/aiq_agent/fastapi_extensions/**,frontends/aiq_api/src/aiq_api/**}
⚙️ CodeRabbit configuration file
{src/aiq_agent/fastapi_extensions/**,frontends/aiq_api/src/aiq_api/**}: Treat API, auth, and job-runner changes as externally visible contracts. Check authorization boundaries,
request tracing, async job lifecycle, websocket reconnect behavior, error responses, and cross-user data isolation.
Require tests for route behavior, access decisions, and job state transitions when those surfaces change.
Files:
frontends/aiq_api/src/aiq_api/jobs/runner.pyfrontends/aiq_api/src/aiq_api/routes/jobs.py
**/*test*.py
📄 CodeRabbit inference engine (CONTRIBUTING.md)
Run pytest for all behavior changes in Python code
Files:
frontends/aiq_api/tests/test_ghost_reaper.py
🧠 Learnings (1)
📚 Learning: 2026-06-14T17:49:00.640Z
Learnt from: torkian
Repo: NVIDIA-AI-Blueprints/aiq PR: 273
File: frontends/aiq_api/tests/test_sse_reconnect_cursor.py:384-401
Timestamp: 2026-06-14T17:49:00.640Z
Learning: When using `unittest.mock.patch` for code that imports dependencies inside functions/generators (e.g., inside `aiq_api.routes.jobs`), don’t patch via an attribute that doesn’t exist on the consuming module. If the generator does `from ..jobs.event_store import EventStore` inside the generator body, then `aiq_api.routes.jobs` will not have an `EventStore` attribute; patch the source class/method in its defining module instead (e.g., `aiq_api.jobs.event_store.EventStore.get_events_async`). Patching `aiq_api.routes.jobs.EventStore...` would raise `AttributeError` because that symbol is not present at module scope.
Applied to files:
frontends/aiq_api/tests/test_ghost_reaper.py
🔇 Additional comments (3)
frontends/aiq_api/src/aiq_api/routes/jobs.py (1)
1204-1224: LGTM!Also applies to: 1230-1253
frontends/aiq_api/src/aiq_api/jobs/runner.py (1)
31-31: LGTM!Also applies to: 194-232, 586-637, 968-971
frontends/aiq_api/tests/test_ghost_reaper.py (1)
270-276: LGTM!
Review follow-up: the worker's terminal SUCCESS path was a read-then-write (check status, then update_job_output), which leaves a TOCTOU window — the reaper could set FAILURE between the read and the write, and the unconditional write would resurrect the job as SUCCESS. Replace it with a single guarded UPDATE (status/output/updated_at WHERE status='running'), matching the reaper's conditional transition, so terminal state is truly first-writer-wins. Output is serialized/encrypted by the new serialize_job_output_for_storage helper (identical bytes to update_job_output) before the atomic write. Also use a small positive lease-refresh interval in the thread test instead of 0 to avoid a zero-interval busy loop. Adds CAS regression tests: the success write flips a running job but is a no-op (returns False, status unchanged) once the job is already terminal. Full frontends/aiq_api suite passes (321); ruff clean. Signed-off-by: Torkian <torkian@mac.com>
|
/ok to test 34f2d3e |
|
@torkian pytest failing |
The success path no longer calls update_job_output; it serializes/encrypts via serialize_job_output_for_storage and then does a conditional write. Retarget the encryption-failure injection to that function so the test still asserts its invariant — an output-write failure marks the job FAILURE, writes no plaintext, and the output assembled the real report (not the output_metadata decoy). Full tests/ suite passes locally (env- gated helm/guardrails/azure suites excluded). Signed-off-by: Torkian <torkian@mac.com>
Strengthen the final-output-failure test: besides checking no plaintext reaches JobStore.update_status, assert the raw-SQL success writer (_write_job_success_if_running_sync) is never called when serialization fails — proving nothing is persisted at all, not just that the ORM path was skipped. Verified the assertion is meaningful (it fails if the writer would run). Signed-off-by: Torkian <torkian@mac.com>
|
Fixed the Pytest failure ( Full |
|
/ok to test 0a2e898 |
|
/merge |
78701dc
into
NVIDIA-AI-Blueprints:release/2.2
2 participants
Closes #317.
Summary
The ghost-job reaper is meant to mark abandoned
RUNNINGjobs asFAILUREwhen a Dask worker crashes or is OOM-killed without raising a Python exception. Its detection query,_find_stale_jobs, was driven fromjob_eventswith anINNER JOIN job_info, so a job that enteredRUNNINGbut had not persisted any events yet produced zero rows and was invisible to the reaper.That is exactly the failure the reaper's own docstring says it catches: the runner marks a job
RUNNINGbefore its first event is stored (andBatchingEventStorebatches writes, widening the window). A worker crash/OOM in that pre-first-event window left the job stuck inrunningforever — status polling never reached a terminal state, and the job permanently counted against the active-job cap.Fix
Drive the query from
job_info LEFT JOIN job_eventsand useCOALESCE(MAX(je.created_at), ji.updated_at)as the staleness clock:GHOST_JOB_TIMEOUT_SECONDS).job_info.updated_at(set when the job enteredRUNNING), so a never-emitted ghost past the timeout is now reaped.job_infotable (the query's new primary table).I verified in a real SQLite DB that both timestamp columns compare correctly —
job_info.updated_at(YYYY-MM-DD HH:MM:SS.ffffff) andjob_events.created_at(YYYY-MM-DD HH:MM:SS) are both text and order correctly against SQLitedatetime(); in Postgres both aretimestamptzand compare natively.Tests
New
frontends/aiq_api/tests/test_ghost_reaper.pyruns against a real SQLite database so the SQL executes exactly as in production:test_zero_event_running_job_past_timeout_is_reaped— the regression: a zero-event RUNNING ghost is now reaped.test_zero_event_running_job_within_timeout_is_not_reaped— a freshly-started zero-event job is left alone.test_running_job_with_stale_last_event_is_reaped/..._recent_event_is_not_reaped— existing event-based behavior preserved.test_non_running_job_is_never_reaped— only RUNNING jobs are candidates.test_missing_tables_returns_empty— fresh deployment, nothing to reap.test_mixed_fleet_reaps_only_ghosts— realistic mix returns only the two genuine ghosts.The three zero-event assertions fail against the old INNER JOIN and pass with the fix.
Test plan
pytest frontends/aiq_api/tests/test_ghost_reaper.py— 7/7 passfrontends/aiq_api/tests/suite — 311 passed, no regressionsruff check+ruff format --checkcleanSummary by CodeRabbit
RUNNINGjobs can be detected and reaped even when no job events have been persisted, using the latest heartbeat/lease timestamp fallback.RUNNING, and recordingGhostJobTimeoutwhen applicable.SUCCESSpersistence with a compare-and-set so it won’t overwrite terminal states.