fix(TaskProducer): bounded queue of pending futures

[bmckerry]

Previously, TaskProducer added pending futures to a set with the assumption that collect_futures() would be routinely called. This isn't necessarily true (as a user could theoretically use TaskProducer outside of a task function), so this changes _pending_futures from a set to a deque with a maximum length.

[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 a615652. Configure here.}

[cursor]

Evicted futures skip completion checks

High Severity

With _pending_futures capped at TASK_PRODUCER_MAX_PENDING_FUTURES, each append past the limit drops the oldest tracked future. A single activation that produces more than that many messages before collect_futures() runs can finish as success even though some Kafka produces were never included in pending_futures or awaited.

 

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

[bmckerry]

Quick Claude search says we don't have any tasks in Sentry that produce more than ~1000 messages per task, so I'm going to assume 10,000 is a safe cap.

[evanh]

Should this return an error, just in case someone does decide to produce >10k messages?

[bmckerry]

Discussed in slack, I opened a PR to record a metric for queue size that we can alert on instead.