feat(pubsub): add CancellationSharer and HedgingScheduler

[feywind]

Overview

This PR is Layer 3 of 7 in Stack #9467, which implements Publish Hedging for @google-cloud/pubsub to mitigate high-percentile (p99/p99.9) publish tail latencies by sending backup (hedged) publish RPCs when an in-flight batch takes longer than a configurable hedgeDelay threshold.

Stack Structure

1. feat(gax): support cancelling API calls via CallOptions.signal #9462 - feat(gax): support AbortSignal cancellation in CallOptions and createApiCall
2. feat(pubsub): add HedgingOptions validation and token bucket #9463 - feat(pubsub): add HedgingOptions validation and HedgingTokenBucket
3. feat(pubsub): add CancellationSharer and HedgingScheduler #9464 - feat(pubsub): add CancellationSharer and HedgingScheduler (this PR)
4. feat(pubsub): add OpenTelemetry events and logging for hedged publish #9465 - feat(pubsub): add OpenTelemetry events and logging for hedged publish
5. feat(pubsub): integrate publish hedging into MessageQueue #9466 - feat(pubsub): integrate publish hedging into MessageQueue
6. test(pubsub): add unit tests for publish hedging queue integration #9635 - test(pubsub): add unit tests for publish hedging queue integration
7. test(pubsub): add publish hedging system tests #9636 - test(pubsub): add publish hedging system tests

---

Summary of Changes in This PR

This layer implements the core hedging state machine (CancellationSharer) and the event-driven single-timer FIFO hedging queue (HedgingScheduler) using Duration arithmetic:

• CancellationSharer (handwritten/pubsub/src/publisher/hedging.ts):
  • Coordinates the original attempt (attempt 0) and any hedged attempts (attempt 1..N) for a single publish batch.
  • First Success Wins: Whichever attempt resolves first marks the batch as done, refills the token bucket once (onSuccess()), aborts all other in-flight attempts via their AbortControllers, and resolves coordinator.promise with {response, wasHedged, successfulAttempt}.
  • Failure Handling:
    • If a hedged attempt (attempt > 0) fails while the original attempt (attempt 0 with GAX retries) is still running, the error is discarded so the batch can still succeed.
    • If the original attempt (attempt 0) fails terminally (permanent error or after exhausting GAX retries), all remaining attempts are aborted and coordinator.promise rejects immediately.
• HedgedRequest & HedgingScheduler (handwritten/pubsub/src/publisher/hedging.ts):
  • Maintains a chronological FIFO queue of HedgedRequest items and a single setTimeout handle (queueProcessingTimer) scheduled for the head item's sendAfter Duration.
  • When an item becomes due and the batch is still in flight with positive remainingTimeout:
    • Computes attemptTimeout = atMost(remainingTimeout, MAX_HEDGE_ATTEMPT_TIMEOUT) (capped at 10s).
    • Attempts tokenBucket.tryAcquireHedgeToken(). If acquired, enqueues the next hedge check (attemptNumber + 1 at now.add(hedgeDelay)) for multiple hedging and invokes startHedgedAttempt(attemptNumber, attemptTimeout).
    • If no token is available, invokes onRateLimited and does not re-enqueue the batch.
• Unit Tests (handwritten/pubsub/test/publisher/hedging.ts):
  • Tests for CancellationSharer (original wins, hedged wins, hedged failure ignored while original runs, original failure aborts hedged attempts) and HedgingScheduler (multiple hedging at hedgeDelay intervals, 10s/remainingTimeout capping, early completion skipping).

[gemini-code-assist]

Code Review

This pull request introduces publish hedging orchestration and scheduling for the Pub/Sub publisher, adding the CancellationSharer to coordinate attempts and the HedgingScheduler to manage the timing of hedged requests. The review feedback highlights a critical issue where the coordinator prematurely aborts hedged attempts and rejects the promise if the original attempt fails, rather than waiting for all attempts to fail. Additionally, the reviewer recommends rejecting the promise with a cancellation error when cancel() is called to prevent hangs, adding an early return in scheduleFirstHedge if the coordinator is already done, optimizing processQueue by using the pre-calculated now duration, and updating the unit tests to reflect the corrected failure behavior.