test: refactor record_waiting_tasks_metric metric

[carlosthe19916]

Fixes: #7940

Addresses: https://redhat.atlassian.net/browse/PULP-2150

Problem

KEDA scales pulp-workers from OpenTelemetry waiting_tasks (emitted by Redis worker record_waiting_tasks_metric). Today that is roughly (WAITING + RUNNING) − workers, which ignores resource locks. Under exclusive contention, queue depth ≫ useful parallelism → over-scale → DB pool saturation → crash loops / orphan Redis locks (incident 2026-07-24).

Fix direction: the metric

pulpcore/pulpcore/tasking/redis_worker.py

Line 321 in dfd517c

def record_waiting_tasks_metric(self):

must reflect how many tasks can usefully run in parallel (resource-aware demand), not raw unfinished count.

Solution

• First commit: creates the test to verify the metric count is the expected one -> expected to fail
• Second commit: refactors the logic for counting the metrics -> tests expected to pass
• Third commit: add more tests

📜 Checklist

• Commits are cleanly separated with meaningful messages (simple features and bug fixes should be squashed to one commit)
• A changelog entry or entries has been added for any significant changes
• Follows the Pulp policy on AI Usage
• (For new features) - User documentation and test coverage has been added

See: Pull Request Walkthrough

[dkliban]

The pulpcore-manager shell subprocess isn't needed here. Since the functional tests already run inside a Django-initialized environment, you can import and call the function directly:

from pulpcore.tasking.redis_worker import count_waiting_tasks_for_metric

def _read_waiting_tasks_metric_value(num_workers=1):
    return count_waiting_tasks_for_metric(num_workers)

This avoids spawning a full Django process six times across the test suite and makes failures easier to debug (you get a real traceback instead of stderr from a subprocess).

[carlosthe19916]

Done, the last commit is importing directly count_waiting_tasks_for_metric

[dkliban]

Missing test case: shared-shared concurrency. The three tests here cover exclusive-exclusive, two independent exclusive lanes, and exclusive-blocks-shared — but none verify that multiple tasks needing only shared access to the same resource are all counted as parallel lanes.

That's the core correctness property of shared locks. If someone accidentally added shared resources to taken_exclusive in the algorithm, only this test would catch it.

Something like: dispatch N tasks all using shared_resources=[same_resource] (no exclusive holder), and assert the metric rises by N, not 1.

[carlosthe19916]

Done, the last test is adding this scenario. Thank for pointing that out!

[carlosthe19916]

@dkliban I addressed your comments! Thanks for the review!