fix(api): recover scans stranded in executing by a dead worker#12008
fix(api): recover scans stranded in executing by a dead worker#12008Subhro-ai wants to merge 2 commits into
Conversation
github-actions
Bot
added
documentation
component/api
review-django-migrations
|
✅ No Conflicts No conflict markers, and the branch merges cleanly into its base. |
📝 WalkthroughWalkthroughChangesThe PR adds a periodic Celery task that identifies stale scans, marks eligible scans failed, and dispatches blocked provider queues. It adds configurable thresholds, migration-based scheduling, recovery exclusions, tests, changelog content, and troubleshooting guidance. Scan cleanup and scheduling
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant CeleryBeat
participant cleanup_stale_scans_task
participant cleanup_stale_scans
participant ScanDatabase
participant ProviderQueueDispatcher
CeleryBeat->>cleanup_stale_scans_task: trigger every 15 minutes
cleanup_stale_scans_task->>cleanup_stale_scans: run cleanup
cleanup_stale_scans->>ScanDatabase: lock and inspect stale scan
cleanup_stale_scans->>ScanDatabase: mark scan and task result failed
cleanup_stale_scans->>ProviderQueueDispatcher: dispatch next queued scan
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 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: 6
🤖 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 `@api/src/backend/config/django/base.py`:
- Around line 324-327: Validate SCAN_INACTIVITY_THRESHOLD_MINUTES and
SCAN_STALE_THRESHOLD_MINUTES during Django configuration startup, rejecting zero
or negative values and any stale threshold lower than the inactivity threshold.
Raise the configuration error through the existing startup validation mechanism
before watchdog processing begins.
In `@api/src/backend/tasks/jobs/scan_cleanup.py`:
- Around line 62-66: Replace the cross-tenant Scan.all_objects query in the
cleanup flow with one tenant-scoped cleanup task per validated tenant. In each
task, wrap the Scan.objects query and cleanup operations in
rls_transaction(tenant_id), preserving the executing-state filter and
task-runner select_related without using MainRouter.admin_db.
- Around line 86-91: Update the _fail_stale_scan call in the scan cleanup flow
so revocation is enabled whenever task_result exists, including when worker is
absent. Preserve non-terminating behavior for the no-worker case by passing
terminate=False, while retaining termination for scans with a live worker.
In `@api/src/backend/tasks/tasks.py`:
- Around line 703-707: Update cleanup_stale_scans_task to use the mandated
RLSTask Celery base while preserving its existing task name, queue, and
cleanup_stale_scans delegation. Ensure the task declaration inherits from
RLSTask so tenant-scoped scan access runs with RLS awareness.
In `@api/src/backend/tasks/tests/test_scan_cleanup.py`:
- Around line 34-85: Update the tenant-scoped fixture helpers that create and
update Scan and Task records, including the setup around _create_queued_scan, to
execute all ORM queries inside rls_transaction(tenant.id). Use the normal Scan
manager instead of Scan.all_objects where RLS applies, and keep any subsequent
refreshes or assertions within the matching tenant transaction.
In `@docs/troubleshooting.mdx`:
- Line 205: Update the recovery estimate in the scan-cleanup-stale-scans
documentation to account for the task’s 15-minute scheduling interval: describe
the default recovery window as approximately 30–45 minutes after the last scan
activity, rather than within 30 minutes of worker death. Preserve the
surrounding explanation and manual workaround.
🪄 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: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 22ab13c1-f392-422c-954d-fa79964e97ae
📒 Files selected for processing (9)
api/changelog.d/scan-stuck-queued.fixed.mdapi/src/backend/api/migrations/0098_scan_cleanup_periodic_task.pyapi/src/backend/config/django/base.pyapi/src/backend/tasks/jobs/orphan_recovery.pyapi/src/backend/tasks/jobs/scan_cleanup.pyapi/src/backend/tasks/tasks.pyapi/src/backend/tasks/tests/test_orphan_recovery.pyapi/src/backend/tasks/tests/test_scan_cleanup.pydocs/troubleshooting.mdx
| SCAN_INACTIVITY_THRESHOLD_MINUTES = env.int("SCAN_INACTIVITY_THRESHOLD_MINUTES", 30) | ||
| SCAN_STALE_THRESHOLD_MINUTES = env.int( | ||
| "SCAN_STALE_THRESHOLD_MINUTES", 2880 | ||
| ) # 48h |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Validate cleanup thresholds during startup.
Zero or negative values make current scans immediately stale, while a stale threshold below the inactivity threshold reverses the documented recovery policy. Reject invalid configuration before the watchdog can fail active scans.
Proposed validation
SCAN_STALE_THRESHOLD_MINUTES = env.int(
"SCAN_STALE_THRESHOLD_MINUTES", 2880
) # 48h
+
+if SCAN_INACTIVITY_THRESHOLD_MINUTES <= 0:
+ raise ValueError("SCAN_INACTIVITY_THRESHOLD_MINUTES must be positive")
+if SCAN_STALE_THRESHOLD_MINUTES < SCAN_INACTIVITY_THRESHOLD_MINUTES:
+ raise ValueError(
+ "SCAN_STALE_THRESHOLD_MINUTES must be greater than or equal to "
+ "SCAN_INACTIVITY_THRESHOLD_MINUTES"
+ )📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| SCAN_INACTIVITY_THRESHOLD_MINUTES = env.int("SCAN_INACTIVITY_THRESHOLD_MINUTES", 30) | |
| SCAN_STALE_THRESHOLD_MINUTES = env.int( | |
| "SCAN_STALE_THRESHOLD_MINUTES", 2880 | |
| ) # 48h | |
| SCAN_INACTIVITY_THRESHOLD_MINUTES = env.int("SCAN_INACTIVITY_THRESHOLD_MINUTES", 30) | |
| SCAN_STALE_THRESHOLD_MINUTES = env.int( | |
| "SCAN_STALE_THRESHOLD_MINUTES", 2880 | |
| ) # 48h | |
| if SCAN_INACTIVITY_THRESHOLD_MINUTES <= 0: | |
| raise ValueError("SCAN_INACTIVITY_THRESHOLD_MINUTES must be positive") | |
| if SCAN_STALE_THRESHOLD_MINUTES < SCAN_INACTIVITY_THRESHOLD_MINUTES: | |
| raise ValueError( | |
| "SCAN_STALE_THRESHOLD_MINUTES must be greater than or equal to " | |
| "SCAN_INACTIVITY_THRESHOLD_MINUTES" | |
| ) |
🤖 Prompt for 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.
In `@api/src/backend/config/django/base.py` around lines 324 - 327, Validate
SCAN_INACTIVITY_THRESHOLD_MINUTES and SCAN_STALE_THRESHOLD_MINUTES during Django
configuration startup, rejecting zero or negative values and any stale threshold
lower than the inactivity threshold. Raise the configuration error through the
existing startup validation mechanism before watchdog processing begins.
| executing_scans = list( | ||
| Scan.all_objects.using(MainRouter.admin_db) | ||
| .filter(state=StateChoices.EXECUTING) | ||
| .select_related("task__task_runner_task") | ||
| ) |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy lift
Do not bypass RLS with a cross-tenant scan query.
This loads all tenants’ scans through the admin database without rls_transaction(tenant_id). Dispatch one tenant-scoped cleanup task per validated tenant and query Scan.objects inside its RLS transaction.
As per path instructions, “Any query against tenant-scoped models outside a ViewSet (including Celery tasks) MUST be wrapped in rls_transaction(tenant_id).”
🤖 Prompt for 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.
In `@api/src/backend/tasks/jobs/scan_cleanup.py` around lines 62 - 66, Replace the
cross-tenant Scan.all_objects query in the cleanup flow with one tenant-scoped
cleanup task per validated tenant. In each task, wrap the Scan.objects query and
cleanup operations in rls_transaction(tenant_id), preserving the executing-state
filter and task-runner select_related without using MainRouter.admin_db.
Sources: Coding guidelines, Path instructions
| if _fail_stale_scan( | ||
| scan, | ||
| task_result, | ||
| reason, | ||
| revoke=worker is not None, | ||
| inactivity_cutoff=recheck_inactivity_cutoff, |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the target file and the helper that performs stale-scan failure handling.
git ls-files 'api/src/backend/tasks/jobs/scan_cleanup.py' 'api/src/backend/tasks/jobs/*' | sed -n '1,20p'
echo '--- scan_cleanup.py (relevant range) ---'
nl -ba api/src/backend/tasks/jobs/scan_cleanup.py | sed -n '1,260p'
echo '--- search for _fail_stale_scan and revoke_task ---'
rg -n "_fail_stale_scan|revoke_task|terminate=False|task_result\.task_id|worker is not None" api/src/backend/tasks/jobs -SRepository: prowler-cloud/prowler
Length of output: 1189
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
path = Path("api/src/backend/tasks/jobs/scan_cleanup.py")
text = path.read_text().splitlines()
for start, end in [(1, 260)]:
for i in range(start-1, min(end, len(text))):
print(f"{i+1:4d}: {text[i]}")
PY
echo '--- matching symbols ---'
python3 - <<'PY'
from pathlib import Path
import re
root = Path("api/src/backend/tasks/jobs")
patterns = [r"_fail_stale_scan", r"revoke_task", r"terminate=False", r"task_result\.task_id", r"worker is not None"]
for path in root.rglob("*.py"):
try:
text = path.read_text()
except Exception:
continue
hits = [p for p in patterns if re.search(p, text)]
if hits:
print(path)
for p in hits:
print(" ", p)
PYRepository: prowler-cloud/prowler
Length of output: 10597
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
for rel in [
"api/src/backend/tasks/jobs/orphan_recovery.py",
"api/src/backend/tasks/jobs/attack_paths/cleanup.py",
]:
path = Path(rel)
print(f"--- {rel} ---")
text = path.read_text().splitlines()
for i, line in enumerate(text, 1):
if "def revoke_task" in line or "terminate=False" in line or "worker is not None" in line or "task_result.task_id" in line:
start = max(1, i-20)
end = min(len(text), i+60)
for j in range(start, end+1):
print(f"{j:4d}: {text[j-1]}")
print()
PYRepository: prowler-cloud/prowler
Length of output: 46638
api/src/backend/tasks/jobs/scan_cleanup.py:86-91 — Revoke stale task IDs even when no worker is recorded. revoke=worker is not None skips Celery revocation in the no-worker branch, leaving the old task_id eligible for later broker redelivery after the scan is marked failed and the queue is drained. Revoke whenever task_result exists, using terminate=False when no live worker is known.
🤖 Prompt for 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.
In `@api/src/backend/tasks/jobs/scan_cleanup.py` around lines 86 - 91, Update the
_fail_stale_scan call in the scan cleanup flow so revocation is enabled whenever
task_result exists, including when worker is absent. Preserve non-terminating
behavior for the no-worker case by passing terminate=False, while retaining
termination for scans with a live worker.
| @shared_task(name="scan-cleanup-stale-scans", queue="celery") | ||
| def cleanup_stale_scans_task(): | ||
| """Periodic watchdog: fail scans stranded in `executing` by a dead worker | ||
| and dispatch the next queued scan the dead scan was blocking.""" | ||
| return cleanup_stale_scans() |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Run the watchdog with the required RLSTask base.
This periodic task accesses tenant-scoped scan data but omits the mandated RLS-aware Celery base.
-@shared_task(name="scan-cleanup-stale-scans", queue="celery")
+@shared_task(
+ base=RLSTask,
+ name="scan-cleanup-stale-scans",
+ queue="celery",
+)As per coding guidelines, tasks must “inherit from RLSTask base class for Celery tasks.”
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| @shared_task(name="scan-cleanup-stale-scans", queue="celery") | |
| def cleanup_stale_scans_task(): | |
| """Periodic watchdog: fail scans stranded in `executing` by a dead worker | |
| and dispatch the next queued scan the dead scan was blocking.""" | |
| return cleanup_stale_scans() | |
| `@shared_task`( | |
| base=RLSTask, | |
| name="scan-cleanup-stale-scans", | |
| queue="celery", | |
| ) | |
| def cleanup_stale_scans_task(): | |
| """Periodic watchdog: fail scans stranded in `executing` by a dead worker | |
| and dispatch the next queued scan the dead scan was blocking.""" | |
| return cleanup_stale_scans() |
🤖 Prompt for 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.
In `@api/src/backend/tasks/tasks.py` around lines 703 - 707, Update
cleanup_stale_scans_task to use the mandated RLSTask Celery base while
preserving its existing task name, queue, and cleanup_stale_scans delegation.
Ensure the task declaration inherits from RLSTask so tenant-scoped scan access
runs with RLS awareness.
Sources: Coding guidelines, Path instructions
| scan = Scan.objects.create( | ||
| tenant_id=tenant.id, | ||
| provider=provider, | ||
| name="Running scan", | ||
| trigger=Scan.TriggerChoices.MANUAL, | ||
| state=StateChoices.EXECUTING, | ||
| started_at=started_at or datetime.now(tz=UTC), | ||
| ) | ||
|
|
||
| task_result = None | ||
| if worker is not None: | ||
| task_result = TaskResult.objects.create( | ||
| task_id=str(scan.id), | ||
| task_name="scan-perform", | ||
| status=task_status, | ||
| worker=worker, | ||
| ) | ||
| task = Task.objects.create( | ||
| id=task_result.task_id, | ||
| task_runner_task=task_result, | ||
| tenant_id=tenant.id, | ||
| ) | ||
| scan.task = task | ||
| scan.save(update_fields=["task_id"]) | ||
|
|
||
| if updated_at is not None: | ||
| # `updated_at` uses auto_now, so bypass it with a queryset update. | ||
| Scan.all_objects.filter(id=scan.id).update(updated_at=updated_at) | ||
| scan.updated_at = updated_at | ||
|
|
||
| return scan, task_result | ||
|
|
||
| def _create_queued_scan(self, tenant, provider): | ||
| """Create an AVAILABLE scan whose task is QUEUED behind the active one.""" | ||
| task_result = TaskResult.objects.create( | ||
| task_id=str(uuid.uuid4()), | ||
| task_name="scan-perform", | ||
| status="QUEUED", | ||
| ) | ||
| task = Task.objects.create( | ||
| id=task_result.task_id, | ||
| task_runner_task=task_result, | ||
| tenant_id=tenant.id, | ||
| ) | ||
| scan = Scan.objects.create( | ||
| tenant_id=tenant.id, | ||
| provider=provider, | ||
| name="Queued scan", | ||
| trigger=Scan.TriggerChoices.MANUAL, | ||
| state=StateChoices.AVAILABLE, | ||
| task=task, | ||
| ) |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy lift
Exercise tenant-scoped fixtures under RLS.
These helpers query Scan and Task without rls_transaction(tenant_id), so the suite can bypass the same RLS contract required in production. Wrap tenant-scoped creation, updates, and subsequent refreshes/assertions in the matching tenant transaction; avoid all_objects where the normal RLS manager is sufficient.
As per path instructions, “Any query against tenant-scoped models outside a ViewSet … MUST be wrapped in rls_transaction(tenant_id).”
🤖 Prompt for 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.
In `@api/src/backend/tasks/tests/test_scan_cleanup.py` around lines 34 - 85,
Update the tenant-scoped fixture helpers that create and update Scan and Task
records, including the setup around _create_queued_scan, to execute all ORM
queries inside rls_transaction(tenant.id). Use the normal Scan manager instead
of Scan.all_objects where RLS applies, and keep any subsequent refreshes or
assertions within the matching tenant transaction.
Sources: Coding guidelines, Path instructions
a63870d to
1c4f170
Compare
|
Thanks for the automated review. This task is intentionally modeled on the existing Applied
Kept as-is, matching the existing watchdog precedent
Happy to adjust any of these if you'd prefer a different convention for the scan reaper specifically. |
josema-xyz
left a comment
josema-xyz
left a comment
There was a problem hiding this comment.
Thanks for taking this on. The approach of failing a stuck scan and then releasing the provider queue makes sense, and the tests and documentation make the change easy to follow.
I left three inline comments about:
- avoiding a false worker-death result from a single short ping;
- retrying the queue release if it fails after the scan is marked as failed;
- confirming this covers #12007, since the report uses 5.22.0 and this cleanup only checks
EXECUTINGscans.
I do not think the CodeRabbit suggestions about RLS, RLSTask, or revoking tasks without a recorded worker apply here. The threshold validation is reasonable, but I do not see it as blocking.
Once the inline points are addressed, this should be ready for another review.
| # Release the provider queue the dead scan was blocking. Runs after | ||
| # the FAILED state is committed so the next dispatch no longer sees | ||
| # this scan as the active one. | ||
| _drain_provider_queue(str(scan.tenant_id), str(scan.provider_id)) |
There was a problem hiding this comment.
If this call fails, the scan is already FAILED. Future cleanup runs only select EXECUTING scans, so they will not try to start the queued scan again. Could this leave the provider queue stuck after a temporary error?
There was a problem hiding this comment.
Fixed. Added a self-healing second pass (_drain_pending_provider_queues) that runs on every cleanup and releases any provider queue still holding a QUEUED scan with no active scan. So if the inline release fails transiently after the scan is already FAILED, the next run retries it — recovery no longer depends on re-selecting the now-terminal scan. _dispatch_next_queued_provider_scan re-checks under lock and no-ops when a scan is genuinely active, so healthy serialization is untouched.
| if (tr := _task_result_for(scan)) and tr.worker | ||
| } | ||
| # Ping each distinct worker once; errors are treated as "alive" upstream. | ||
| worker_alive = {worker: is_worker_alive(worker) for worker in workers} |
There was a problem hiding this comment.
This check runs before marking the scan as failed and stopping its task, but is_worker_alive() only waits one second and returns False when the worker does not reply in time. A busy worker could be treated as dead. Since #11986 already handles this with retries and an unknown state, should we use the same logic here?
There was a problem hiding this comment.
Good call. Switched to the same bounded-retry ping used for Attack Paths in #11986 (_ping_workers): it pings with exponential backoff (5s/10s/20s) over up to 3 attempts and returns an "unknown" state when the control bus fails. A scan is only failed when its worker is in the confirmed-unresponsive set — a busy or briefly-unreachable worker is preserved and retried on the next run.
|
|
||
| executing_scans = list( | ||
| Scan.all_objects.using(MainRouter.admin_db) | ||
| .filter(state=StateChoices.EXECUTING) |
There was a problem hiding this comment.
This only checks EXECUTING scans. A task can be lost before the scan reaches that state, while _get_dispatched_provider_scan still treats it as active and blocks the queue. Also, this PR uses Prowler version 5.22, before #11848 added this queue. Can we confirm this PR fixes the reported case?
There was a problem hiding this comment.
Two parts:
- Task lost before
executing: the candidate set now also includesavailable/scheduledscans whose task is still in a dispatched state (PENDING/STARTED/PROGRESS) — the same thing_get_dispatched_provider_scantreats as active — so a task lost before it reachesexecutingis reaped too. A scan merely waiting in the broker queue (no recorded worker, nostarted_at) is preserved, so this never fails a not-yet-started scan. - Kick off a new Scan of a provider - Stuck in Queued status #12007 / 5.22.0: the reporter’s 5.22.0 predates the per-provider queue (fix(api): queue provider scans when one is active #11848, v5.33.0), so on that version a "Queued" (
available) scan that never starts is a worker/broker throughput problem rather than this DB queue. On current master the queue exists, and a worker dying mid-scan strands theexecutingscan and leaves every following scan for that provider inQueued— the reported symptom ("other jobs sometimes start, but the queued one never does"). This watchdog targets that path. I couldn’t run the API suite locally (no uv/Postgres on my box), so the new unit tests are the main verification — happy to add any scenario you’d like.
Since prowler-cloud#11848 scans run one-at-a-time per provider: a new scan waits in `available` ("Queued") until the active scan's task finishes and its `finally` block dispatches the next queued scan. `scan-perform` uses `acks_late=False` and is excluded from orphan recovery, so a worker that dies mid-scan (OOM, hard time-limit kill, node failure) never runs that `finally`. The scan is stranded in `executing` and, because it still counts as the provider's active scan, every queued scan for that provider stays "Queued" forever (prowler-cloud#12007). Add a periodic `scan-cleanup-stale-scans` task that reaps such scans - confirmed dead by pinging the recorded worker - marks them `failed`, revokes the lost task, and dispatches the next queued scan so the provider's queue drains. It mirrors `cleanup_stale_attack_paths_scans` for the main `Scan` model and never re-runs the dead scan, which would duplicate findings. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Address review feedback on the scan-cleanup watchdog: - Replace the single 1s worker ping with the bounded-retry + unknown-state ping logic from prowler-cloud#11986, so a busy worker is not misjudged as dead and its running scan is never failed by mistake. - Also reap scans lost before reaching `executing` (available/scheduled with a still-dispatched task), which block the provider queue the same way. - Add a self-healing second pass that releases any provider queue holding a QUEUED scan but no active scan, so a queue drain dropped by a transient error is retried on the next run instead of stranding the queue. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1c4f170 to
0586609
Compare
|
@josema-xyz thanks for the thorough review. Pushed a commit addressing all three inline points:
Replied inline with the details (including the #12007 / 5.22.0 confirmation). Left the CodeRabbit RLS/ |
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 `@api/src/backend/tasks/jobs/scan_cleanup.py`:
- Around line 130-140: Verify PostgreSQL index coverage for the periodic filters
in _stale_scan_candidates and _drain_pending_provider_queues, including
Scan.state and the joined TaskResult.status fields used by
task__task_runner_task. Inspect existing model metadata and migrations, run
EXPLAIN analysis for both queries, and add only necessary indexes through a
migration if coverage is missing.
🪄 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: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 6406fed0-dcf9-4497-b601-ff8468abc724
📒 Files selected for processing (9)
api/changelog.d/scan-stuck-queued.fixed.mdapi/src/backend/api/migrations/0098_scan_cleanup_periodic_task.pyapi/src/backend/config/django/base.pyapi/src/backend/tasks/jobs/orphan_recovery.pyapi/src/backend/tasks/jobs/scan_cleanup.pyapi/src/backend/tasks/tasks.pyapi/src/backend/tasks/tests/test_orphan_recovery.pyapi/src/backend/tasks/tests/test_scan_cleanup.pydocs/troubleshooting.mdx
| return list( | ||
| Scan.all_objects.using(MainRouter.admin_db) | ||
| .filter( | ||
| Q(state=StateChoices.EXECUTING) | ||
| | Q( | ||
| state__in=(StateChoices.AVAILABLE, StateChoices.SCHEDULED), | ||
| task__task_runner_task__status__in=DISPATCHED_SCAN_TASK_STATES, | ||
| ) | ||
| ) | ||
| .select_related("task__task_runner_task") | ||
| ) |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🔵 Trivial
Verify indexes support the periodic filter queries.
Both _stale_scan_candidates and _drain_pending_provider_queues filter on Scan.state plus a join through task__task_runner_task__status, run every 15 minutes across all tenants. Worth confirming these columns are indexed as the scans/task_results tables grow, per PostgreSQL indexing guidance for this codebase.
As per path instructions: "Use PostgreSQL indexing and EXPLAIN analysis to optimize slow queries."
#!/bin/bash
# Check for existing indexes/migrations touching Scan.state or TaskResult.status
rg -n "state" api/src/backend/api/migrations -g '*.py' | rg -i "index"
rg -n "class Meta" -A15 api/src/backend/api/models.py | rg -i "index"Also applies to: 343-351
🤖 Prompt for 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.
In `@api/src/backend/tasks/jobs/scan_cleanup.py` around lines 130 - 140, Verify
PostgreSQL index coverage for the periodic filters in _stale_scan_candidates and
_drain_pending_provider_queues, including Scan.state and the joined
TaskResult.status fields used by task__task_runner_task. Inspect existing model
metadata and migrations, run EXPLAIN analysis for both queries, and add only
necessary indexes through a migration if coverage is missing.
Source: Path instructions
There was a problem hiding this comment.
Both _stale_scan_candidates and _drain_pending_provider_queues filter to the small set of non-terminal scans (executing/available/scheduled) — at any moment roughly on the order of the number of active providers — and the TaskResult join is keyed off that small candidate set rather than scanned. There is currently no state-leading index on scans, so the periodic filter does rely on a scan of the table.
Happy to add a partial index on the non-terminal states (which both periodic queries would use) via AddIndexConcurrently, following the existing perf-index migrations. @josema-xyz — would you like that included in this PR, or kept as a focused follow-up?
2 participants
Context
Fix #12007
Since #11848 (v5.33.0) scans run one-at-a-time per provider. When a scan is launched while another is active for the same provider, the new scan is created in the
availablestate (shown as Queued in the UI) and is not dispatched to Celery — it waits until the active scan's task finishes and itsfinallyblock dispatches the next queued scan.scan-performusesacks_late=Falseand is deliberately excluded from orphan recovery, so a worker that dies mid-scan (OOM, hard time-limit kill, node failure) never runs thatfinally. The scan is stranded inexecuting(the scenario already documented in the troubleshooting guide / #6893) and, because it still counts as the provider's active scan, every queued scan for that provider stays "Queued" forever — which is exactly what the reporter saw ("other jobs sometimes start, but the job in the queue never starts").Description
Adds a periodic watchdog task
scan-cleanup-stale-scansthat recovers scans stranded inexecutingby a dead worker, mirroring the existingcleanup_stale_attack_paths_scansfor the mainScanmodel:executingscans (cross-tenant, admin DB) and pings each recorded worker once.SCAN_STALE_THRESHOLD_MINUTES, default 48h, matched to the long hard time-limit);SCAN_INACTIVITY_THRESHOLD_MINUTES, default 30m of no progress heartbeat);started_at;failed(withcompleted_at/duration), marks theTaskResultterminal, and revokes the lost task after commit.It never re-runs the dead scan (that would duplicate findings) — it only fails it and releases the queue. The task is registered as a
PeriodicTask(every 15 min) and added to_SKIP_RECOVERYso the orphan watchdog never re-enqueues it.New settings (env-overridable, no
.envchanges required — same convention as the Attack Paths thresholds):SCAN_INACTIVITY_THRESHOLD_MINUTES(default 30)SCAN_STALE_THRESHOLD_MINUTES(default 2880)Files:
tasks/jobs/scan_cleanup.py— new reaper + queue-draintasks/tasks.py—cleanup_stale_scans_task(queue="celery")tasks/jobs/orphan_recovery.py— skip the new maintenance taskconfig/django/base.py— thresholdsapi/migrations/0098_scan_cleanup_periodic_task.py— PeriodicTask (15 min)tasks/tests/test_scan_cleanup.py— unit teststasks/tests/test_orphan_recovery.py— cover skip of the new taskdocs/troubleshooting.mdx— note automatic recovery now existsapi/changelog.d/scan-stuck-queued.fixed.mdSteps to review
tasks/jobs/scan_cleanup.py. Confirm the detection matrix (alive/dead/no-worker) and that queue drain runs only after the FAILED state commits.cd api && uv run pytest src/backend/tasks/tests/test_scan_cleanup.py src/backend/tasks/tests/test_orphan_recovery.py -qscan-cleanup-stale-scansPeriodicTask:uv run python src/backend/manage.py migrate.docker killthe worker) so the scan is left inexecuting, queue another scan for the same provider (it goes toQueued), then run the task (uv run python src/backend/manage.py shell -c "from tasks.tasks import cleanup_stale_scans_task; print(cleanup_stale_scans_task())") and confirm the stuck scan flips tofailedand the queued scan starts.Checklist
test_scan_cleanup.py) and orphan-recovery skip covered.api/changelog.d/.By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
Summary by CodeRabbit
Bug Fixes
Maintenance
Documentation
Tests