feat(TaskProducer): add metrics on queue size

[bmckerry]

This updates TaskProducer to take a metrics backend, and reports a gauge metric based on its pending_futures queue size.
We can use this to alert if a TaskProducer instance is going to reach its future queue max size, leading to dropped futures.

[cursor]

Cursor Bugbot has reviewed your changes and found 2 potential issues.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

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

[cursor]

Gauge stale after queue cleared

Medium Severity

task.producer.pending.futures is updated only in track_future when a future is appended, but collect_futures clears _pending_futures without recording a gauge. After each activation (including when the queue is empty), dashboards can still show the last non-zero size and trigger false queue-pressure alerts.

Additional Locations (1)

• clients/python/src/taskbroker_client/worker/producer.py#L54-L61

 

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

[cursor]

Producer tag omits other producers

Medium Severity

The gauge uses producer_name but reports len(_pending_futures), a global deque shared by every TaskProducer. Only the instance that called track_future updates its tag, so another producer’s futures can leave that series stale and under-report true queue pressure for that name.

 

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

[bmckerry]

_pending_futures is global out of necessity, but there should generally only be 1 TaskProducer instance per worker replica.