fix(fetch): Undo the claim when submitting to the push pool fails

[enochtangg]

** Description

When the fetch thread claims an activation but cannot hand it to the push pool, the row is left in the Claimed state with nothing holding it. The claim query only selects Pending rows, so nothing re-picks it and it sits until handle_claim_expiration reverts it.

This inflates the SLO metric. pending_activation.max_lag.sec counts Claimed rows, so one stranded activation drives it up until the lease expires. This symptom can be observed consistently in process-segments-push in s4s2: DD link. During this time, throughput, occupancy and AlloyDB latency were all normal, so the metric was tracking a single stuck row rather than real pipeline latency.

When workers briefly go unavailable (due to a deployment), failed pushes each revert their own claim, which is correct, but the retried work fills the push queue to its cap. Activations that then time out submitting to the push queue are not reverted. The fetch thread only logs them.

**Fix

Release the claim when a submit to the push pool fails, for both Timeout and Closed.

• Submit without parking the activation. The fetch thread uses try_send in a retry loop instead of send_async wrapped in a timeout. A failed try_send hands the activation back, so an error proves the push pool never received it.
• Release the rest of the batch. The first failure releases the current activation and everything left in the batch in one query, then stops. Before, a closed queue stranded the rest of the batch, and a timeout made each remaining activation wait out its own timeout.
• ActivationStore::release_claims only releases rows that are still Claimed with the claim_expires_at they were fetched with. Every claim writes a new expiry, so a release can never undo a later claim on the same row. A row already marked Processing is also left alone.

This is safe because only activations that never left the broker are released, so no worker can be running them. A submit failure now costs roughly the fetch backoff instead of the full claim lease. The push thread already reverted inline after a failed gRPC push. That logic moved to ActivationStore::undo_claims, so the push thread gets the same fenced release.

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

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

Want reviews to match your repository better? Bugbot Learning can learn team-specific rules from PR activity. A team admin can enable Learning in the Cursor dashboard.

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

[untitaker]

if we abort early here, then the remaining ids in the batch are still stuck in Claimed, right?

also isn't this fundamentally racy? if the claim expires before we get to this line, we might undo somebody else's claim.

[untitaker]

just tested it and fable was able to find the same thing (and more things i didn't verify)

[enochtangg]

Good catches, tried to fix with the following:

• When a submit fails, the fetch thread now releases the current activation and the rest of the batch in one query.
• Releases also check claim_expires_at now. A release happens if the row still has the expiry we claimed it with, so it can't undo someone else's newer claim. I tested this end to end locally.