Dedupe batch completion across both code paths

[lovitt]

The bug

on_batch_complete can be triggered from two code paths that may observe a batch as complete at the same time:

• on_child_complete — the last child to finish notifies the parent.
• complete — the parent's own post-setup backstop, run at the end of execute.

Only on_child_complete claimed the batch_completion_gid SETNX token before firing on_complete; complete fired it directly. When both observe complete? == true concurrently, on_batch_complete could fire twice.

The fix

Extract the guarded fire into trigger_completion(status) — SETNX the token, run on_complete, clear the token on error so completion can be reattempted — and route both paths through it, so a concurrent backstop and child no longer both fire.

on_batch_node_complete stays outside the guard: it's a distinct per-node event that complete fires once on its own.

Testing

• #trigger_completion: fires when unclaimed, forwards a non-:completed status, no-ops when already claimed, clears the key and re-raises on error.
• #complete: claims the token when complete; defers (no on_complete) when the token is already held, while on_batch_node_complete still propagates.

Scope and limitations

The SETNX guard dedupes the common case: two paths racing to complete a batch that's still in flight. It does not handle Cloud Tasks delivering the same job twice — an uncommon at-least-once redelivery. A job that already completed can arrive after cleanup has wiped the batch state and the token, re-create a phantom slot, and re-fire on_batch_complete. Cloudtasker runs a job every time Cloud Tasks delivers it and never checks whether that job already ran, so workers needing exactly-once completion should keep perform and on_batch_complete idempotent.