Worker stops permanently after peer resets `GetWorkItems` stream

[torosent]

What happened

DurableTaskSchedulerWorker permanently stopped receiving work after DTS reset the GetWorkItems stream:

status = StatusCode.CANCELLED
details = "Received RST_STREAM with error code 8"
Disconnected from ...
No longer listening for work items

The process and container remained healthy, but the scheduler's ConnectedWorkers metric dropped from 1 to 0 until the container was restarted. I observed this twice, after connections lasting roughly 10.5 and 13.7 hours.

There was no container restart, resource pressure, deployment, or Azure service-health event at either disconnect.

Cause

In TaskHubGrpcWorker._async_run_loop, CANCELLED always exits the worker loop:

if error_code == grpc.StatusCode.CANCELLED:
    self._logger.info(f"Disconnected from {self._host_address}")
    break

This also happens for peer-initiated cancellation, not only worker.stop(). The connection is invalidated, but the common reconnect path is skipped.

The current tests use CANCELLED as a convenient way to stop the worker loop, so this case is not covered.

Expected behavior

A peer-initiated CANCELLED should reconnect. The worker should exit only when shutdown was requested.

For example:

if error_code == grpc.StatusCode.CANCELLED and self._shutdown.is_set():
    self._logger.info(f"Disconnected from {self._host_address}")
    break

Otherwise, CANCELLED can continue through the existing connection-reset and retry path.

Environment

• durabletask==1.7.2
• durabletask-azuremanaged==1.7.2
• Python 3.12