Keep activity heartbeats in order when encoding is slow

[parbhatkapila4]

Each activity.heartbeat() call starts its own task to encode and record the details. When encoding actually awaits (a codec waiting on a remote KMS, or external storage), those tasks can finish out of order: a newer heartbeat gets recorded, then an older one lands after it. Core keeps the last recorded details, so a retried activity resumes from the older ones.

The new test shows it on main: the activity heartbeats "first" then "second", the log shows ('second',) recorded before ('first',), and the retry gets "first".

The fix drops the queue. Only the newest heartbeat was ever sent, so the activity now keeps just that one, and each heartbeat task takes a per-activity lock before picking it up. Only one heartbeat encodes and records at a time, and a task that finds nothing left returns. asyncio.Lock wakes waiters in order, so the completion path awaiting last_heartbeat_task still waits for every heartbeat ahead of it. Without the queue, heartbeat() can no longer raise QueueFull either.

[CLAassistant]

All committers have signed the CLA.

[tconley1428]

The issue seems valid, but the fix looks fairly overcomplicated. There's now three layers of queue draining, which shouldn't be necessary. Either simplify the solution, or create an issue instead.

[parbhatkapila4]

Agreed, that was more than it needed.
Now simplified it: the queue is gone, since only the newest heartbeat was ever sent, so the activity just keeps that one. Each heartbeat task takes a per-activity lock before sending, so only one encodes and records at a time. No draining left. The test still fails on main and passes with this, and the rest of test_activity.py passes too.