We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Temporary network connectivity problems cause Worker to stop working
### First check

- [X] I added a descriptive title to this issue.
- [X] I used the GitHub search to find a similar issue and didn't find it.
- [X] I searched the Prefect documentation for this issue.
- [X] I checked that this issue is related to Prefect and not one of its dependencies.

### Bug summary

Network issues seem to get the Worker into a state where it continues to send heartbeats, but stops running flows, resulting in flows getting stuck in "pending" state.

### Reproduction

```python3
Run a worker and wait until it gets "stuck"
```


### Error

```python3
$ k logs prefect-worker-stable-78d664b779-snmln
/usr/local/lib/python3.11/site-packages/prefect/experimental/cli/worker.py:81: ExperimentalFeature: The workers feature is experimental. The interface or behavior may change without warning, we recommend pinning versions to prevent unexpected changes. To disable warnings for this group of experiments, disable PREFECT_EXPERIMENTAL_WARN_WORKERS.
  async with worker_cls(
Worker 'KubernetesWorker 3ee16ee0-5575-4d11-975b-a644c66b2362' started!

Failed the last 3 attempts.  Please check your environment and configuration.
Examples of recent errors:

Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/h2/connection.py", line 224, in 
process_input
    func, target_state = self._transitions[(self.state, input_)]
                         ~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^
KeyError: (<ConnectionState.CLOSED: 3>, <ConnectionInputs.SEND_HEADERS: 0>)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/httpcore/_async/http2.py", line 
106, in handle_async_request
    await self._send_request_headers(request=request, stream_id=stream_id)
  File "/usr/local/lib/python3.11/site-packages/httpcore/_async/http2.py", line 
203, in _send_request_headers
    self._h2_state.send_headers(stream_id, headers, end_stream=end_stream)
  File "/usr/local/lib/python3.11/site-packages/h2/connection.py", line 766, in 
send_headers
    self.state_machine.process_input(ConnectionInputs.SEND_HEADERS)
  File "/usr/local/lib/python3.11/site-packages/h2/connection.py", line 228, in 
process_input
    raise ProtocolError(
h2.exceptions.ProtocolError: Invalid input ConnectionInputs.SEND_HEADERS in 
state ConnectionState.CLOSED

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/httpx/_transports/default.py", 
line 60, in map_httpcore_exceptions
    yield
  File "/usr/local/lib/python3.11/site-packages/httpx/_transports/default.py", 
line 353, in handle_async_request
    resp = await self._pool.handle_async_request(req)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File 
"/usr/local/lib/python3.11/site-packages/httpcore/_async/connection_pool.py", 
line 253, in handle_async_request
    raise exc
  File 
"/usr/local/lib/python3.11/site-packages/httpcore/_async/connection_pool.py", 
line 237, in handle_async_request
    response = await connection.handle_async_request(request)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/httpcore/_async/connection.py", 
line 90, in handle_async_request
    return await self._connection.handle_async_request(request)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/httpcore/_async/http2.py", line 
142, in handle_async_request
    raise LocalProtocolError(exc)  # pragma: nocover
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
httpcore.LocalProtocolError: Invalid input ConnectionInputs.SEND_HEADERS in 
state ConnectionState.CLOSED

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/prefect/utilities/services.py", 
line 46, in critical_service_loop
    await workload()
  File 
"/usr/local/lib/python3.11/site-packages/prefect/experimental/workers/base.py", 
line 372, in get_and_submit_flow_runs
    runs_response = await self._get_scheduled_flow_runs()
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File 
"/usr/local/lib/python3.11/site-packages/prefect/experimental/workers/base.py", 
line 440, in _get_scheduled_flow_runs
    await self._client.get_scheduled_flow_runs_for_work_pool(
  File 
"/usr/local/lib/python3.11/site-packages/prefect/client/orchestration.py", line 
2269, in get_scheduled_flow_runs_for_work_pool
    response = await self._client.post(
               ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/httpx/_client.py", line 1848, in
post
    return await self.request(
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/httpx/_client.py", line 1533, in
request
    return await self.send(request, auth=auth, 
follow_redirects=follow_redirects)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
^
  File "/usr/local/lib/python3.11/site-packages/prefect/client/base.py", line 
243, in send
    response = await self._send_with_retry(
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/client/base.py", line 
189, in _send_with_retry
    response = await request()
               ^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/httpx/_client.py", line 1620, in
send
    response = await self._send_handling_auth(
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/httpx/_client.py", line 1648, in
_send_handling_auth
    response = await self._send_handling_redirects(
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/httpx/_client.py", line 1685, in
_send_handling_redirects
    response = await self._send_single_request(request)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/httpx/_client.py", line 1722, in
_send_single_request
    response = await transport.handle_async_request(request)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/httpx/_transports/default.py", 
line 352, in handle_async_request
    with map_httpcore_exceptions():
  File "/usr/local/lib/python3.11/contextlib.py", line 155, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/usr/local/lib/python3.11/site-packages/httpx/_transports/default.py", 
line 77, in map_httpcore_exceptions
    raise mapped_exc(message) from exc
httpx.LocalProtocolError: Invalid input ConnectionInputs.SEND_HEADERS in state 
ConnectionState.CLOSED
```


### Versions

```Text
Image:         prefecthq/prefect-dev:main-python3.11-kubernetes
    Image ID:      docker.io/prefecthq/prefect-dev@sha256:0b2bd51e56b1a61b8570f5973192270a7afbf49bf808c9e125168f048c25cdb0
```


### Additional context

_No response_

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/prefect/utilities/services.py`
Content:
```
1 import sys
2 from collections import deque
3 from traceback import format_exception
4 from types import TracebackType
5 from typing import Callable, Coroutine, Deque, Tuple
6 
7 import anyio
8 import httpx
9 
10 from prefect.utilities.collections import distinct
11 from prefect.utilities.math import clamped_poisson_interval
12 
13 
14 async def critical_service_loop(
15     workload: Callable[..., Coroutine],
16     interval: float,
17     memory: int = 10,
18     consecutive: int = 3,
19     printer: Callable[..., None] = print,
20     run_once: bool = False,
21     jitter_range: float = None,
22 ):
23     """
24     Runs the given `workload` function on the specified `interval`, while being
25     forgiving of intermittent issues like temporary HTTP errors.  If more than a certain
26     number of `consecutive` errors occur, print a summary of up to `memory` recent
27     exceptions to `printer`, then exit.
28 
29     Args:
30         workload: the function to call
31         interval: how frequently to call it
32         memory: how many recent errors to remember
33         consecutive: how many consecutive errors must we see before we exit
34         printer: a `print`-like function where errors will be reported
35         run_once: if set, the loop will only run once then return
36         jitter_range: if set, the interval will be a random variable (rv) drawn from
37             a clamped Poisson distribution where lambda = interval and the rv is bound
38             between `interval * (1 - range) < rv < interval * (1 + range)`
39     """
40 
41     track_record: Deque[bool] = deque([True] * consecutive, maxlen=consecutive)
42     failures: Deque[Tuple[Exception, TracebackType]] = deque(maxlen=memory)
43 
44     while True:
45         try:
46             await workload()
47 
48             track_record.append(True)
49         except httpx.TransportError as exc:
50             # httpx.TransportError is the base class for any kind of communications
51             # error, like timeouts, connection failures, etc.  This does _not_ cover
52             # routine HTTP error codes (even 5xx errors like 502/503) so this
53             # handler should not be attempting to cover cases where the Prefect server
54             # or Prefect Cloud is having an outage (which will be covered by the
55             # exception clause below)
56             track_record.append(False)
57             failures.append((exc, sys.exc_info()[-1]))
58         except httpx.HTTPStatusError as exc:
59             if exc.response.status_code in (502, 503):
60                 # 502/503 indicate a potential outage of the Prefect server or Prefect
61                 # Cloud, which is likely to be temporary and transient.  Don't quit
62                 # over these unless it is prolonged.
63                 track_record.append(False)
64                 failures.append((exc, sys.exc_info()[-1]))
65             else:
66                 raise
67         except KeyboardInterrupt:
68             return
69 
70         # Decide whether to exit now based on recent history.
71         #
72         # Given some typical background error rate of, say, 1%, we may still observe
73         # quite a few errors in our recent samples, but this is not necessarily a cause
74         # for concern.
75         #
76         # Imagine two distributions that could reflect our situation at any time: the
77         # everything-is-fine distribution of errors, and the everything-is-on-fire
78         # distribution of errors. We are trying to determine which of those two worlds
79         # we are currently experiencing.  We compare the likelihood that we'd draw N
80         # consecutive errors from each.  In the everything-is-fine distribution, that
81         # would be a very low-probability occurrance, but in the everything-is-on-fire
82         # distribution, that is a high-probability occurrance.
83         #
84         # Remarkably, we only need to look back for a small number of consecutive
85         # errors to have reasonable confidence that this is indeed an anomaly.
86         # @anticorrelator and @chrisguidry estimated that we should only need to look
87         # back for 3 consectutive errors.
88         if not any(track_record):
89             # We've failed enough times to be sure something is wrong, the writing is
90             # on the wall.  Let's explain what we've seen and exit.
91             printer(
92                 f"\nFailed the last {consecutive} attempts.  "
93                 "Please check your environment and configuration."
94             )
95 
96             printer("Examples of recent errors:\n")
97 
98             failures_by_type = distinct(
99                 reversed(failures),
100                 key=lambda pair: type(pair[0]),  # Group by the type of exception
101             )
102             for exception, traceback in failures_by_type:
103                 printer("".join(format_exception(None, exception, traceback)))
104                 printer()
105             return
106 
107         if run_once:
108             return
109 
110         if jitter_range is not None:
111             sleep = clamped_poisson_interval(interval, clamping_factor=jitter_range)
112         else:
113             sleep = interval
114 
115         await anyio.sleep(sleep)
116
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/prefect/utilities/services.py b/src/prefect/utilities/services.py
--- a/src/prefect/utilities/services.py
+++ b/src/prefect/utilities/services.py
@@ -64,8 +64,6 @@
                 failures.append((exc, sys.exc_info()[-1]))
             else:
                 raise
-        except KeyboardInterrupt:
-            return
 
         # Decide whether to exit now based on recent history.
         #
@@ -89,7 +87,7 @@
             # We've failed enough times to be sure something is wrong, the writing is
             # on the wall.  Let's explain what we've seen and exit.
             printer(
-                f"\nFailed the last {consecutive} attempts.  "
+                f"\nFailed the last {consecutive} attempts. "
                 "Please check your environment and configuration."
             )
 
@@ -102,7 +100,8 @@
             for exception, traceback in failures_by_type:
                 printer("".join(format_exception(None, exception, traceback)))
                 printer()
-            return
+
+            raise RuntimeError("Service exceeded error threshold.")
 
         if run_once:
             return

{"golden_diff": "diff --git a/src/prefect/utilities/services.py b/src/prefect/utilities/services.py\n--- a/src/prefect/utilities/services.py\n+++ b/src/prefect/utilities/services.py\n@@ -64,8 +64,6 @@\n                 failures.append((exc, sys.exc_info()[-1]))\n             else:\n                 raise\n-        except KeyboardInterrupt:\n-            return\n \n         # Decide whether to exit now based on recent history.\n         #\n@@ -89,7 +87,7 @@\n             # We've failed enough times to be sure something is wrong, the writing is\n             # on the wall.  Let's explain what we've seen and exit.\n             printer(\n-                f\"\\nFailed the last {consecutive} attempts.  \"\n+                f\"\\nFailed the last {consecutive} attempts. \"\n                 \"Please check your environment and configuration.\"\n             )\n \n@@ -102,7 +100,8 @@\n             for exception, traceback in failures_by_type:\n                 printer(\"\".join(format_exception(None, exception, traceback)))\n                 printer()\n-            return\n+\n+            raise RuntimeError(\"Service exceeded error threshold.\")\n \n         if run_once:\n             return\n", "issue": "Temporary network connectivity problems cause Worker to stop working\n### First check\n\n- [X] I added a descriptive title to this issue.\n- [X] I used the GitHub search to find a similar issue and didn't find it.\n- [X] I searched the Prefect documentation for this issue.\n- [X] I checked that this issue is related to Prefect and not one of its dependencies.\n\n### Bug summary\n\nNetwork issues seem to get the Worker into a state where it continues to send heartbeats, but stops running flows, resulting in flows getting stuck in \"pending\" state.\n\n### Reproduction\n\n```python3\nRun a worker and wait until it gets \"stuck\"\n```\n\n\n### Error\n\n```python3\n$ k logs prefect-worker-stable-78d664b779-snmln\r\n/usr/local/lib/python3.11/site-packages/prefect/experimental/cli/worker.py:81: ExperimentalFeature: The workers feature is experimental. The interface or behavior may change without warning, we recommend pinning versions to prevent unexpected changes. To disable warnings for this group of experiments, disable PREFECT_EXPERIMENTAL_WARN_WORKERS.\r\n  async with worker_cls(\r\nWorker 'KubernetesWorker 3ee16ee0-5575-4d11-975b-a644c66b2362' started!\r\n\r\nFailed the last 3 attempts.  Please check your environment and configuration.\r\nExamples of recent errors:\r\n\r\nTraceback (most recent call last):\r\n  File \"/usr/local/lib/python3.11/site-packages/h2/connection.py\", line 224, in \r\nprocess_input\r\n    func, target_state = self._transitions[(self.state, input_)]\r\n                         ~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^\r\nKeyError: (<ConnectionState.CLOSED: 3>, <ConnectionInputs.SEND_HEADERS: 0>)\r\n\r\nDuring handling of the above exception, another exception occurred:\r\n\r\nTraceback (most recent call last):\r\n  File \"/usr/local/lib/python3.11/site-packages/httpcore/_async/http2.py\", line \r\n106, in handle_async_request\r\n    await self._send_request_headers(request=request, stream_id=stream_id)\r\n  File \"/usr/local/lib/python3.11/site-packages/httpcore/_async/http2.py\", line \r\n203, in _send_request_headers\r\n    self._h2_state.send_headers(stream_id, headers, end_stream=end_stream)\r\n  File \"/usr/local/lib/python3.11/site-packages/h2/connection.py\", line 766, in \r\nsend_headers\r\n    self.state_machine.process_input(ConnectionInputs.SEND_HEADERS)\r\n  File \"/usr/local/lib/python3.11/site-packages/h2/connection.py\", line 228, in \r\nprocess_input\r\n    raise ProtocolError(\r\nh2.exceptions.ProtocolError: Invalid input ConnectionInputs.SEND_HEADERS in \r\nstate ConnectionState.CLOSED\r\n\r\nDuring handling of the above exception, another exception occurred:\r\n\r\nTraceback (most recent call last):\r\n  File \"/usr/local/lib/python3.11/site-packages/httpx/_transports/default.py\", \r\nline 60, in map_httpcore_exceptions\r\n    yield\r\n  File \"/usr/local/lib/python3.11/site-packages/httpx/_transports/default.py\", \r\nline 353, in handle_async_request\r\n    resp = await self._pool.handle_async_request(req)\r\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \r\n\"/usr/local/lib/python3.11/site-packages/httpcore/_async/connection_pool.py\", \r\nline 253, in handle_async_request\r\n    raise exc\r\n  File \r\n\"/usr/local/lib/python3.11/site-packages/httpcore/_async/connection_pool.py\", \r\nline 237, in handle_async_request\r\n    response = await connection.handle_async_request(request)\r\n               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"/usr/local/lib/python3.11/site-packages/httpcore/_async/connection.py\", \r\nline 90, in handle_async_request\r\n    return await self._connection.handle_async_request(request)\r\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"/usr/local/lib/python3.11/site-packages/httpcore/_async/http2.py\", line \r\n142, in handle_async_request\r\n    raise LocalProtocolError(exc)  # pragma: nocover\r\n    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\nhttpcore.LocalProtocolError: Invalid input ConnectionInputs.SEND_HEADERS in \r\nstate ConnectionState.CLOSED\r\n\r\nThe above exception was the direct cause of the following exception:\r\n\r\nTraceback (most recent call last):\r\n  File \"/usr/local/lib/python3.11/site-packages/prefect/utilities/services.py\", \r\nline 46, in critical_service_loop\r\n    await workload()\r\n  File \r\n\"/usr/local/lib/python3.11/site-packages/prefect/experimental/workers/base.py\", \r\nline 372, in get_and_submit_flow_runs\r\n    runs_response = await self._get_scheduled_flow_runs()\r\n                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \r\n\"/usr/local/lib/python3.11/site-packages/prefect/experimental/workers/base.py\", \r\nline 440, in _get_scheduled_flow_runs\r\n    await self._client.get_scheduled_flow_runs_for_work_pool(\r\n  File \r\n\"/usr/local/lib/python3.11/site-packages/prefect/client/orchestration.py\", line \r\n2269, in get_scheduled_flow_runs_for_work_pool\r\n    response = await self._client.post(\r\n               ^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"/usr/local/lib/python3.11/site-packages/httpx/_client.py\", line 1848, in\r\npost\r\n    return await self.request(\r\n           ^^^^^^^^^^^^^^^^^^^\r\n  File \"/usr/local/lib/python3.11/site-packages/httpx/_client.py\", line 1533, in\r\nrequest\r\n    return await self.send(request, auth=auth, \r\nfollow_redirects=follow_redirects)\r\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n^\r\n  File \"/usr/local/lib/python3.11/site-packages/prefect/client/base.py\", line \r\n243, in send\r\n    response = await self._send_with_retry(\r\n               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"/usr/local/lib/python3.11/site-packages/prefect/client/base.py\", line \r\n189, in _send_with_retry\r\n    response = await request()\r\n               ^^^^^^^^^^^^^^^\r\n  File \"/usr/local/lib/python3.11/site-packages/httpx/_client.py\", line 1620, in\r\nsend\r\n    response = await self._send_handling_auth(\r\n               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"/usr/local/lib/python3.11/site-packages/httpx/_client.py\", line 1648, in\r\n_send_handling_auth\r\n    response = await self._send_handling_redirects(\r\n               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"/usr/local/lib/python3.11/site-packages/httpx/_client.py\", line 1685, in\r\n_send_handling_redirects\r\n    response = await self._send_single_request(request)\r\n               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"/usr/local/lib/python3.11/site-packages/httpx/_client.py\", line 1722, in\r\n_send_single_request\r\n    response = await transport.handle_async_request(request)\r\n               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"/usr/local/lib/python3.11/site-packages/httpx/_transports/default.py\", \r\nline 352, in handle_async_request\r\n    with map_httpcore_exceptions():\r\n  File \"/usr/local/lib/python3.11/contextlib.py\", line 155, in __exit__\r\n    self.gen.throw(typ, value, traceback)\r\n  File \"/usr/local/lib/python3.11/site-packages/httpx/_transports/default.py\", \r\nline 77, in map_httpcore_exceptions\r\n    raise mapped_exc(message) from exc\r\nhttpx.LocalProtocolError: Invalid input ConnectionInputs.SEND_HEADERS in state \r\nConnectionState.CLOSED\n```\n\n\n### Versions\n\n```Text\nImage:         prefecthq/prefect-dev:main-python3.11-kubernetes\r\n    Image ID:      docker.io/prefecthq/prefect-dev@sha256:0b2bd51e56b1a61b8570f5973192270a7afbf49bf808c9e125168f048c25cdb0\n```\n\n\n### Additional context\n\n_No response_\n", "before_files": [{"content": "import sys\nfrom collections import deque\nfrom traceback import format_exception\nfrom types import TracebackType\nfrom typing import Callable, Coroutine, Deque, Tuple\n\nimport anyio\nimport httpx\n\nfrom prefect.utilities.collections import distinct\nfrom prefect.utilities.math import clamped_poisson_interval\n\n\nasync def critical_service_loop(\n    workload: Callable[..., Coroutine],\n    interval: float,\n    memory: int = 10,\n    consecutive: int = 3,\n    printer: Callable[..., None] = print,\n    run_once: bool = False,\n    jitter_range: float = None,\n):\n    \"\"\"\n    Runs the given `workload` function on the specified `interval`, while being\n    forgiving of intermittent issues like temporary HTTP errors.  If more than a certain\n    number of `consecutive` errors occur, print a summary of up to `memory` recent\n    exceptions to `printer`, then exit.\n\n    Args:\n        workload: the function to call\n        interval: how frequently to call it\n        memory: how many recent errors to remember\n        consecutive: how many consecutive errors must we see before we exit\n        printer: a `print`-like function where errors will be reported\n        run_once: if set, the loop will only run once then return\n        jitter_range: if set, the interval will be a random variable (rv) drawn from\n            a clamped Poisson distribution where lambda = interval and the rv is bound\n            between `interval * (1 - range) < rv < interval * (1 + range)`\n    \"\"\"\n\n    track_record: Deque[bool] = deque([True] * consecutive, maxlen=consecutive)\n    failures: Deque[Tuple[Exception, TracebackType]] = deque(maxlen=memory)\n\n    while True:\n        try:\n            await workload()\n\n            track_record.append(True)\n        except httpx.TransportError as exc:\n            # httpx.TransportError is the base class for any kind of communications\n            # error, like timeouts, connection failures, etc.  This does _not_ cover\n            # routine HTTP error codes (even 5xx errors like 502/503) so this\n            # handler should not be attempting to cover cases where the Prefect server\n            # or Prefect Cloud is having an outage (which will be covered by the\n            # exception clause below)\n            track_record.append(False)\n            failures.append((exc, sys.exc_info()[-1]))\n        except httpx.HTTPStatusError as exc:\n            if exc.response.status_code in (502, 503):\n                # 502/503 indicate a potential outage of the Prefect server or Prefect\n                # Cloud, which is likely to be temporary and transient.  Don't quit\n                # over these unless it is prolonged.\n                track_record.append(False)\n                failures.append((exc, sys.exc_info()[-1]))\n            else:\n                raise\n        except KeyboardInterrupt:\n            return\n\n        # Decide whether to exit now based on recent history.\n        #\n        # Given some typical background error rate of, say, 1%, we may still observe\n        # quite a few errors in our recent samples, but this is not necessarily a cause\n        # for concern.\n        #\n        # Imagine two distributions that could reflect our situation at any time: the\n        # everything-is-fine distribution of errors, and the everything-is-on-fire\n        # distribution of errors. We are trying to determine which of those two worlds\n        # we are currently experiencing.  We compare the likelihood that we'd draw N\n        # consecutive errors from each.  In the everything-is-fine distribution, that\n        # would be a very low-probability occurrance, but in the everything-is-on-fire\n        # distribution, that is a high-probability occurrance.\n        #\n        # Remarkably, we only need to look back for a small number of consecutive\n        # errors to have reasonable confidence that this is indeed an anomaly.\n        # @anticorrelator and @chrisguidry estimated that we should only need to look\n        # back for 3 consectutive errors.\n        if not any(track_record):\n            # We've failed enough times to be sure something is wrong, the writing is\n            # on the wall.  Let's explain what we've seen and exit.\n            printer(\n                f\"\\nFailed the last {consecutive} attempts.  \"\n                \"Please check your environment and configuration.\"\n            )\n\n            printer(\"Examples of recent errors:\\n\")\n\n            failures_by_type = distinct(\n                reversed(failures),\n                key=lambda pair: type(pair[0]),  # Group by the type of exception\n            )\n            for exception, traceback in failures_by_type:\n                printer(\"\".join(format_exception(None, exception, traceback)))\n                printer()\n            return\n\n        if run_once:\n            return\n\n        if jitter_range is not None:\n            sleep = clamped_poisson_interval(interval, clamping_factor=jitter_range)\n        else:\n            sleep = interval\n\n        await anyio.sleep(sleep)\n", "path": "src/prefect/utilities/services.py"}], "after_files": [{"content": "import sys\nfrom collections import deque\nfrom traceback import format_exception\nfrom types import TracebackType\nfrom typing import Callable, Coroutine, Deque, Tuple\n\nimport anyio\nimport httpx\n\nfrom prefect.utilities.collections import distinct\nfrom prefect.utilities.math import clamped_poisson_interval\n\n\nasync def critical_service_loop(\n    workload: Callable[..., Coroutine],\n    interval: float,\n    memory: int = 10,\n    consecutive: int = 3,\n    printer: Callable[..., None] = print,\n    run_once: bool = False,\n    jitter_range: float = None,\n):\n    \"\"\"\n    Runs the given `workload` function on the specified `interval`, while being\n    forgiving of intermittent issues like temporary HTTP errors.  If more than a certain\n    number of `consecutive` errors occur, print a summary of up to `memory` recent\n    exceptions to `printer`, then exit.\n\n    Args:\n        workload: the function to call\n        interval: how frequently to call it\n        memory: how many recent errors to remember\n        consecutive: how many consecutive errors must we see before we exit\n        printer: a `print`-like function where errors will be reported\n        run_once: if set, the loop will only run once then return\n        jitter_range: if set, the interval will be a random variable (rv) drawn from\n            a clamped Poisson distribution where lambda = interval and the rv is bound\n            between `interval * (1 - range) < rv < interval * (1 + range)`\n    \"\"\"\n\n    track_record: Deque[bool] = deque([True] * consecutive, maxlen=consecutive)\n    failures: Deque[Tuple[Exception, TracebackType]] = deque(maxlen=memory)\n\n    while True:\n        try:\n            await workload()\n\n            track_record.append(True)\n        except httpx.TransportError as exc:\n            # httpx.TransportError is the base class for any kind of communications\n            # error, like timeouts, connection failures, etc.  This does _not_ cover\n            # routine HTTP error codes (even 5xx errors like 502/503) so this\n            # handler should not be attempting to cover cases where the Prefect server\n            # or Prefect Cloud is having an outage (which will be covered by the\n            # exception clause below)\n            track_record.append(False)\n            failures.append((exc, sys.exc_info()[-1]))\n        except httpx.HTTPStatusError as exc:\n            if exc.response.status_code in (502, 503):\n                # 502/503 indicate a potential outage of the Prefect server or Prefect\n                # Cloud, which is likely to be temporary and transient.  Don't quit\n                # over these unless it is prolonged.\n                track_record.append(False)\n                failures.append((exc, sys.exc_info()[-1]))\n            else:\n                raise\n\n        # Decide whether to exit now based on recent history.\n        #\n        # Given some typical background error rate of, say, 1%, we may still observe\n        # quite a few errors in our recent samples, but this is not necessarily a cause\n        # for concern.\n        #\n        # Imagine two distributions that could reflect our situation at any time: the\n        # everything-is-fine distribution of errors, and the everything-is-on-fire\n        # distribution of errors. We are trying to determine which of those two worlds\n        # we are currently experiencing.  We compare the likelihood that we'd draw N\n        # consecutive errors from each.  In the everything-is-fine distribution, that\n        # would be a very low-probability occurrance, but in the everything-is-on-fire\n        # distribution, that is a high-probability occurrance.\n        #\n        # Remarkably, we only need to look back for a small number of consecutive\n        # errors to have reasonable confidence that this is indeed an anomaly.\n        # @anticorrelator and @chrisguidry estimated that we should only need to look\n        # back for 3 consectutive errors.\n        if not any(track_record):\n            # We've failed enough times to be sure something is wrong, the writing is\n            # on the wall.  Let's explain what we've seen and exit.\n            printer(\n                f\"\\nFailed the last {consecutive} attempts. \"\n                \"Please check your environment and configuration.\"\n            )\n\n            printer(\"Examples of recent errors:\\n\")\n\n            failures_by_type = distinct(\n                reversed(failures),\n                key=lambda pair: type(pair[0]),  # Group by the type of exception\n            )\n            for exception, traceback in failures_by_type:\n                printer(\"\".join(format_exception(None, exception, traceback)))\n                printer()\n\n            raise RuntimeError(\"Service exceeded error threshold.\")\n\n        if run_once:\n            return\n\n        if jitter_range is not None:\n            sleep = clamped_poisson_interval(interval, clamping_factor=jitter_range)\n        else:\n            sleep = interval\n\n        await anyio.sleep(sleep)\n", "path": "src/prefect/utilities/services.py"}]}