Publish one Pub/Sub request per flush

[BinaryFiddler]

Summary

AsyncPubSubPublisher already batches in an asyncio queue. With BatchSettings(max_messages=1), the high-level client still created one commit thread and one publish RPC per message.

This change calls the generated GAPIC client once per asyncio flush. The public API and flush timing are unchanged: messages already queued together share a request, without adding a batching delay.

Results

Microbenchmark: real _sync_flush, RPC stubbed at the same boundary, 4,000 × 1.2 KB messages, three runs.

Messages per flush	CPU/message, before → after	Threads/message
1	74 → 22 µs	1.01 → 0
13	71 → 3.4 µs	1.00 → 0

Production averages about 15.2k attempts/s across these publishers. Most flushes still contain one message, so the steady-state gain is removing per-message wrapper and thread work; bursts batch naturally.

Other fixes

• stop() drains the queue even if the flush task never starts, retries messages requeued during the drain, and records residual messages as shutdown_dropped.
• _stopping prevents the Python 3.11 asyncio.wait_for cancellation race from hanging shutdown.
• Cancelled and Unknown now match the generated client's transient retry set.
• Adds queue_depth and messages_per_request; publish counters remain per message.

Risk

One RPC now shares an outcome across its batch. Transient failures requeue the batch; permanent failures drop it. Seven-day production data shows ~0.17 RetryError/s, ~0.12 TimeoutError/s, and no permanent error tag.

messages_per_request.count measures logical GAPIC publish calls, not extra wire attempts made by internal retries.

Review follow-ups before merge

• Fix the byte guard to check the prospective serialized request size. It currently allows a message to push a batch past Pub/Sub's 10 MB request limit.
• Make the shutdown wording/implementation explicit: the 30-second check stops starting new flushes, but does not bound an in-flight publish or executor wait.
• Remove or justify partial-response recovery. The API does not document that a short successful response means “prefix accepted, tail rejected.”

Verification and rollout

CI is green for async unit tests, Python quality, integration tests, docs, UI, and Rust. The author also exercised burst, paced, immediate-stop, and missing-topic cases against a Pub/Sub emulator.

After rollout, verify attempt rate is stable, failure rate does not rise, queue depth stays near zero, and shutdown_dropped stays zero. Client pooling remains out of scope.