Harden gRPC worker and client against silent disconnects

[berndverst]

Summary

Hardens the gRPC worker and client against silent disconnects and stale transport state when the Durable Task Scheduler (DTS) backend is replaced, drained, or otherwise leaves callers latched onto a bad HTTP/2 path.

Before this change, a worker or long-lived client could stay pinned to a half-open connection, stale sub-channel state, stale DNS result, or stale routing affinity for the lifetime of the process. In practice that meant a backend replacement or graceful drain could leave callers idle until restart even though the service had recovered.

Motivation / root causes

The main failure modes were:

1. HelloAsync on the worker connect path had no deadline, so a half-open handshake could hang forever and prevent the reconnect loop from making progress.
2. The gRPC channel was effectively process-lifetime state. Once the transport was wedged on stale DNS, SocketsHttpHandler state, or backend routing affinity, nothing forced a fresh channel.
3. Worker reconnect used a fixed delay, so a fleet could reconnect in lockstep against a backend that was still warming up.
4. Some transport-adjacent failures were either logged generically (Unauthenticated) or not surfaced clearly at all (graceful server stream end).
5. The silent-disconnect timeout and health-ping cadence were misaligned, which made healthy-but-idle streams harder to reason about and increased the chance of pointless reconnects.
6. Client-side recreate cooldown math depended on a raw double -> long conversion of TimeSpan, which was not well-defined for extreme values.

How this change addresses them

Worker (src/Worker/Grpc)

• Add a 30 second HelloAsync deadline (configurable via GrpcDurableTaskWorkerOptions.SetHelloDeadline()) so a half-open handshake cannot stall reconnect forever.
• Treat Unavailable, hello DeadlineExceeded, silent stream disconnects, and empty graceful drains as transport failures that increment a recreate counter.
• Use capped exponential backoff with full jitter instead of a fixed reconnect delay.
• Log Unauthenticated distinctly.
• Raise the silent-disconnect timeout to 120 seconds and log HealthPing at Trace.
• Handle graceful server stream closes explicitly via Logs.StreamEndedByPeer.
• Keep channel recreation on the worker side, but simplify the Azure Managed worker cache path now that the cache key already includes WorkerId: replace that worker's cached channel, guard against stale callbacks, and defer disposal of the previous channel. This keeps the worker logic focused on the realistic case instead of modeling peer-worker contention as a default scenario.

Client (src/Client/Grpc)

The client does not have a central reconnect loop, so recovery is implemented around the call path instead:

• ChannelRecreatingCallInvoker observes unary RPC outcomes and recreates the channel after repeated transport failures.
• Long-poll DeadlineExceeded calls are excluded from failure counting.
• Recreate stays off the caller-visible failure path via a single-flight background recreate plus cooldown.
• MinRecreateInterval conversion to stopwatch ticks now saturates safely for extreme values.
• The Azure Managed client-side shared-cache recreate flow is preserved, including peer-client winner reuse and deferred disposal, because multiple clients in one process realistically can share the same cached channel.

Azure Managed cache wiring (src/{Worker,Client}/AzureManaged)

• Worker side: keep per-worker cache replacement simple and scoped by WorkerId.
• Client side: keep the richer shared-cache coordination logic because peer-client contention is realistic there.
• Both paths continue to defer disposal of the old channel so in-flight RPCs can drain.

Tests and validation

• Add behavior-level gRPC worker recreation tests.
• Add behavior-level gRPC client recreation tests.
• Add worker Azure Managed coverage for cache replacement and stale recreate callbacks.
• Add a worker test for the UTC DateTime.MaxValue hello-deadline clamp.
• Stabilize the flaky WorkItemStreamConsumer heartbeat-reset timing test that was failing CI.

What this prevents

• Workers hanging indefinitely on a half-open handshake.
• Long-lived workers or clients staying pinned to dead routing targets after backend replacement.
• Slow, opaque recovery from graceful server drains.
• Reconnect storms caused by fixed-delay retry loops.
• Generic/noisy diagnostics for auth and stream-liveness issues.
• Undefined cooldown behavior for extreme client recreate intervals.

PR comment follow-ups

• Simplified the worker-side Azure Managed recreate logic after confirming worker cache keys include WorkerId, so peer-worker coordination is not a realistic default scenario.
• Kept the richer peer-aware recreate logic on the client side, where a shared cached channel is realistic.
• Addressed the current review follow-ups on worker-test cleanup and client cooldown overflow handling.

Breaking change

No breaking change.

[Copilot]

Pull request overview

This PR hardens the Durable Task gRPC worker and client against silent disconnects and wedged HTTP/2 connections by adding bounded handshakes, jittered reconnect backoff, and channel recreation paths (including AzureManaged channel-cache CAS swaps with deferred disposal).

Changes:

• Worker: add Hello deadline, silent-disconnect window changes + HealthPing trace logging, jittered exponential reconnect backoff, and channel recreation signaling.
• Client: introduce a CallInvoker wrapper that triggers channel recreation after consecutive transport failures, with cooldown/single-flight safeguards and long-poll exclusions.
• AzureManaged extensions: wire recreator callbacks through the shared ConcurrentDictionary<string, Lazy<GrpcChannel>> cache with CAS swap + deferred disposal.

Reviewed changes

Copilot reviewed 16 out of 16 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
test/Worker/Grpc.Tests/ReconnectBackoffTests.cs	Adds unit tests for full-jitter exponential reconnect backoff behavior.
test/Worker/Grpc.Tests/GrpcDurableTaskWorkerOptionsInternalTests.cs	Adds tests for internal worker option defaults and SetChannelRecreator validation.
src/Worker/Grpc/ReconnectBackoff.cs	New helper for capped exponential backoff with jitter.
src/Worker/Grpc/ProcessorExitReason.cs	New enum to signal processor exit reason to the outer worker loop.
src/Worker/Grpc/Logs.cs	Adds new structured log events for timeout/auth/channel recreation/backoff/HealthPing observability.
src/Worker/Grpc/Internal/InternalOptionsExtensions.cs	Adds internal extension method to set a channel recreator callback.
src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs	Adds internal options for deadlines, silent-disconnect timeout, backoff, and recreation thresholds.
src/Worker/Grpc/GrpcDurableTaskWorker.cs	Wraps processor execution in a loop and attempts channel recreation when requested.
src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs	Implements Hello deadline, failure tracking, backoff, silent disconnect signaling, and HealthPing trace logging.
src/Worker/AzureManaged/DurableTaskSchedulerWorkerExtensions.cs	Adds cached-channel recreation with CAS swap and deferred channel disposal.
src/Client/Grpc/Logs.cs	Adds client-side channel recreation logs.
src/Client/Grpc/Internal/InternalOptionsExtensions.cs	Adds internal extension method to set a client channel recreator callback.
src/Client/Grpc/GrpcDurableTaskClientOptions.cs	Adds internal client options for recreation threshold and cooldown interval.
src/Client/Grpc/GrpcDurableTaskClient.cs	Wraps the call invoker with a recreating invoker when enabled; owns lifecycle for Address-created channels.
src/Client/Grpc/ChannelRecreatingCallInvoker.cs	New CallInvoker wrapper that recreates channels after consecutive transport failures.
src/Client/AzureManaged/DurableTaskSchedulerClientExtensions.cs	Adds cached-channel recreation with CAS swap and deferred channel disposal for AzureManaged clients.

Comments suppressed due to low confidence (1)

src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs:427

• ProcessWorkItemsAsync can enter a tight loop if the server completes the work-item stream normally (ReadAllAsync finishes without throwing and without any cancellation). In that case, tokenSource is not cancelled, the cancellation check is false, and the outer while-loop restarts another ReadAllAsync on the same completed stream. Consider treating normal stream completion as a disconnect (e.g., return after the await-foreach completes, or break/throw so the caller reconnects) to avoid spinning and to re-establish the stream.

            while (!cancellation.IsCancellationRequested)
            {
                await foreach (P.WorkItem workItem in stream.ResponseStream.ReadAllAsync(tokenSource.Token))
                {
                    if (silentDisconnectEnabled)
                    {
                        timeoutSource.CancelAfter(silentDisconnectTimeout);
                    }

                    if (!firstMessageObserved)
                    {
                        firstMessageObserved = true;
                        onFirstMessage?.Invoke();
                    }

                    if (workItem.RequestCase == P.WorkItem.RequestOneofCase.OrchestratorRequest)
                    {
                        this.RunBackgroundTask(
                            workItem,
                            () => this.OnRunOrchestratorAsync(
                                workItem.OrchestratorRequest,
                                workItem.CompletionToken,
                                cancellation),
                            cancellation);
                    }
                    else if (workItem.RequestCase == P.WorkItem.RequestOneofCase.ActivityRequest)
                    {
                        this.RunBackgroundTask(
                            workItem,
                            () => this.OnRunActivityAsync(
                                workItem.ActivityRequest,
                                workItem.CompletionToken,
                                cancellation),
                            cancellation);
                    }
                    else if (workItem.RequestCase == P.WorkItem.RequestOneofCase.EntityRequest)
                    {
                        this.RunBackgroundTask(
                            workItem,
                            () => this.OnRunEntityBatchAsync(workItem.EntityRequest.ToEntityBatchRequest(), cancellation),
                            cancellation);
                    }
                    else if (workItem.RequestCase == P.WorkItem.RequestOneofCase.EntityRequestV2)
                    {
                        workItem.EntityRequestV2.ToEntityBatchRequest(
                            out EntityBatchRequest batchRequest,
                            out List<P.OperationInfo> operationInfos);

                        this.RunBackgroundTask(
                             workItem,
                             () => this.OnRunEntityBatchAsync(
                                batchRequest,
                                cancellation,
                                workItem.CompletionToken,
                                operationInfos),
                             cancellation);
                    }
                    else if (workItem.RequestCase == P.WorkItem.RequestOneofCase.HealthPing)
                    {
                        // Health pings are heartbeat-only signals from the backend; the silent-disconnect
                        // timer reset above is the actionable behavior. Logging at Trace allows operators
                        // to confirm liveness without flooding info-level telemetry.
                        this.Logger.ReceivedHealthPing();
                    }
                    else
                    {
                        this.Logger.UnexpectedWorkItemType(workItem.RequestCase.ToString());
                    }
                }

                if (tokenSource.IsCancellationRequested || tokenSource.Token.IsCancellationRequested)
                {
                    // The token has cancelled, this means either:
                    // 1. The broader 'cancellation' was triggered, return here to start a graceful shutdown.
                    // 2. The timeoutSource was triggered, return here to trigger a reconnect to the backend.
                    if (!cancellation.IsCancellationRequested)
                    {
                        // Since the cancellation came from the timeout, log a warning.
                        this.Logger.ConnectionTimeout();
                        onSilentDisconnect?.Invoke();
                    }

                    return;
                }
            }

[Copilot]

Pull request overview

Copilot reviewed 16 out of 16 changed files in this pull request and generated 3 comments.

[Copilot]

Pull request overview

Copilot reviewed 16 out of 16 changed files in this pull request and generated 3 comments.

[Copilot]

Pull request overview

Copilot reviewed 19 out of 19 changed files in this pull request and generated 4 comments.

[Copilot]

Pull request overview

Copilot reviewed 18 out of 18 changed files in this pull request and generated 5 comments.

Resetting my vote

[Copilot]

Pull request overview

Copilot reviewed 24 out of 24 changed files in this pull request and generated 3 comments.

[Copilot]

Pull request overview

Copilot reviewed 24 out of 24 changed files in this pull request and generated 1 comment.

[Copilot]

Pull request overview

Copilot reviewed 22 out of 22 changed files in this pull request and generated 2 comments.

[Copilot]

Pull request overview

Copilot reviewed 22 out of 22 changed files in this pull request and generated 1 comment.