fix(pubsub/eventhub): cap in-flight health probes at one to prevent goroutine leak

[Umang01-hash]

Problem

Health() runs the Event Hub connectivity probe on a goroutine so probeWithin can return on its own deadline. That design is correct — the Azure SDK's GetEventHubProperties passes context.Background() to the AMQP RPC (mgmt.go) and the reply-wait selects on that context (internal/rpc.go), so the caller's deadline can't bound it; only selecting on our own deadline does.

The cost, as the existing comment notes, is that an abandoned probe stays parked in the SDK call until it returns. Against a broker that accepts a connection but never answers, that call doesn't return promptly — so every health poll stranded another parked goroutine. Health checks are polled continuously (liveness, /.well-known/health), so this is an unbounded goroutine leak on the hot path, and it compounds exactly when the broker is already degraded.

Fix

Cap concurrent probes at one via TryLock. A poll that finds a probe already parked reports errProbeInFlight (the broker is already unresponsive, so reporting down is correct) instead of piling on another abandoned goroutine. The lock is released by the goroutine that ran the probe, once the SDK call finally returns — so at most one goroutine is ever parked, regardless of poll rate.

Behaviour is unchanged in every healthy/normal-failure case: the probe still returns on its 2s deadline; only the pathological "accepts-but-hangs" broker is now bounded.

Test

Test_Health_ParkedProbeIsCappedAtOne parks a probe against a fake that blocks without reading ctx (what the SDK does once a management link exists), then polls Health() repeatedly and asserts the SDK probe runs at most once while one is still parked. Reverting the cap makes the count climb — verified red.

go test ./... -race passes; golangci-lint --new-from-rev reports 0 new issues.

Follow-up to #3649.

[aryanmehrotra]

The goroutine-leak fix itself is correct and the tests behind it are real. My one blocking concern is that the cap changes behaviour in a case the description says it doesn't.

What I verified

• go build, go vet (incl. copylocks, since Client now holds a mutex) — clean
• go test -race -count=1 ./... — passes
• golangci-lint run --new-from-rev=origin/development — 0 issues, as claimed
• Unlocking from a different goroutine than the one that locked is explicitly legal (sync.Mutex: "A locked Mutex is not associated with a particular goroutine"), so that part is sound
• Both new tests are load-bearing — I mutation-tested each in isolation:
  • revert the cap (drop TryLock + Unlock) → Test_Health_ParkedProbeIsCappedAtOne goes red
  • drop only the Unlock → Test_Health_ProbeRecoversAfterParkedCallReturns goes red

The central claim holds: at most one goroutine parks, regardless of poll rate.

Blocking: two concurrent polls against a healthy broker now report DOWN

Behaviour is unchanged in every healthy/normal-failure case; only the pathological "accepts-but-hangs" broker is now bounded.

That isn't quite right. The cap is per-Client, not per-broker-state — a second caller that arrives while a perfectly healthy probe is in flight gets errProbeInFlight and reports StatusDown.

Two simultaneous Health() calls, healthy broker, 100ms probe (well inside the 2s deadline):

branch	result
development	up=2 down=0
this PR	up=1 down=1

Reproducer:

func Test_ConcurrentPollsOnHealthyBroker(t *testing.T) {
	client := newHealthTestClient(t, &mockConsumerClient{
		getPropsFunc: func(context.Context,
			*azeventhubs.GetEventHubPropertiesOptions) (azeventhubs.EventHubProperties, error) {
			time.Sleep(100 * time.Millisecond) // healthy, but not instant
			return azeventhubs.EventHubProperties{PartitionIDs: []string{"0"}}, nil
		},
	})

	var wg sync.WaitGroup
	var up int32
	start := make(chan struct{})

	for range 2 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			<-start
			if client.Health().Status == datasource.StatusUp {
				atomic.AddInt32(&up, 1)
			}
		}()
	}

	close(start)
	wg.Wait()

	require.Equal(t, int32(2), atomic.LoadInt32(&up),
		"a healthy broker must report UP to every concurrent poller")
}

This is reachable in normal operation. Container.Health calls c.PubSub.Health() sequentially (pkg/gofr/container/health.go:35), so there's no intra-request concurrency — but /.well-known/health is an ordinary HTTP handler and Go serves each request on its own goroutine. Overlapping pollers (k8s liveness + readiness, a load balancer plus a monitor) hit this. The window is narrow, since a healthy probe is milliseconds, but the consequence on a liveness probe is a false DOWN and a pod restart — a false negative on the exact path whose job is to avoid them.

Suggested fix: share the in-flight probe rather than rejecting the caller

Let concurrent callers wait on the same probe's result, each still bounded by its own 2s deadline, instead of returning errProbeInFlight immediately. That keeps the goroutine cap exactly as it is — still one SDK call parked, still one goroutine — and removes the false DOWN outright, because a probe that finishes inside the deadline satisfies every waiter. golang.org/x/sync/singleflight is the idiomatic shape; a shared result channel swapped under a small mutex works too and avoids the dependency.

Worth keeping errProbeInFlight for the genuinely pathological case: a caller whose own deadline expires while the shared probe is still parked should still report down.

Minor: worth stating the recovery trade-off

If a parked probe never returns, health now reports DOWN permanently — even after the broker recovers, where pre-fix a fresh probe would have succeeded. Bounded-leak-then-restart is probably the better trade, but since it's a real change in failure behaviour it belongs in the description rather than being left implicit.