https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I am facing the following error aroun...
# ask-marvin
g
@Marvin I am facing the following error around websocket timeouts, suggest a suitable way to reliably prevent this, or work around it to prevent flow run crashes
Copy code
49 results
Severity
Time
Summary
Loading... Scanned up to 4/6/26, 7:11 AM.
2026-04-06 00:21:59.929 IST

Traceback (most recent call last):   File "/opt/venv/lib/python3.13/site-packages/websockets/asyncio/client.py", line 543, in __await_impl__     await self.connection.handshake(
2026-04-06 00:21:59.929 IST
    ...<2 lines>...
2026-04-06 00:21:59.929 IST
    )
2026-04-06 00:21:59.929 IST
  File "/opt/venv/lib/python3.13/site-packages/websockets/asyncio/client.py", line 104, in handshake
2026-04-06 00:21:59.929 IST
    await asyncio.wait(
2026-04-06 00:21:59.929 IST
    ...<2 lines>...
2026-04-06 00:21:59.929 IST
    )
2026-04-06 00:21:59.929 IST
  File "/usr/local/lib/python3.13/asyncio/tasks.py", line 451, in wait
2026-04-06 00:21:59.929 IST
    return await _wait(fs, timeout, return_when, loop)
2026-04-06 00:21:59.929 IST
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2026-04-06 00:21:59.929 IST
  File "/usr/local/lib/python3.13/asyncio/tasks.py", line 537, in _wait
2026-04-06 00:21:59.929 IST
    await waiter
2026-04-06 00:21:59.929 IST
asyncio.exceptions.CancelledError
2026-04-06 00:21:59.929 IST
The above exception was the direct cause of the following exception:
2026-04-06 00:21:59.929 IST

Traceback (most recent call last):   File "/opt/venv/lib/python3.13/site-packages/websockets/asyncio/client.py", line 539, in __await_impl__     async with asyncio_timeout(self.open_timeout):                ~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^   File "/usr/local/lib/python3.13/asyncio/timeouts.py", line 116, in __aexit__     raise TimeoutError from exc_val
2026-04-06 00:21:59.929 IST
TimeoutError
2026-04-06 00:21:59.929 IST
The above exception was the direct cause of the following exception:
2026-04-06 00:21:59.931 IST

Traceback (most recent call last):   File "/opt/venv/lib/python3.13/site-packages/prefect/cli/_utilities.py", line 44, in wrapper     return fn(*args, **kwargs)   File "/opt/venv/lib/python3.13/site-packages/prefect/cli/_types.py", line 159, in sync_fn     return asyncio.run(async_fn(*args, **kwargs))            ~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^   File "/usr/local/lib/python3.13/asyncio/runners.py", line 195, in run     return runner.run(main)            ~~~~~~~~~~^^^^^^   File "/usr/local/lib/python3.13/asyncio/runners.py", line 118, in run     return self._loop.run_until_complete(task)            ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^   File "/usr/local/lib/python3.13/asyncio/base_events.py", line 725, in run_until_complete     return future.result()            ~~~~~~~~~~~~~^^   File "/opt/venv/lib/python3.13/site-packages/prefect/cli/flow_run.py", line 412, in execute     await runner.execute_flow_run(id)   File "/opt/venv/lib/python3.13/site-packages/prefect/runner/runner.py", line 601, in execute_flow_run     async with context:                ^^^^^^^   File "/opt/venv/lib/python3.13/site-packages/prefect/runner/runner.py", line 1571, in __aenter__     self._cancelling_observer = await self._exit_stack.enter_async_context(                                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2026-04-06 00:21:59.931 IST
    ...<6 lines>...
2026-04-06 00:21:59.931 IST
    )
2026-04-06 00:21:59.931 IST
    ^
2026-04-06 00:21:59.931 IST
  File "/usr/local/lib/python3.13/contextlib.py", line 668, in enter_async_context
2026-04-06 00:21:59.931 IST
    result = await _enter(cm)
2026-04-06 00:21:59.931 IST
             ^^^^^^^^^^^^^^^^
2026-04-06 00:21:59.931 IST
  File "/opt/venv/lib/python3.13/site-packages/prefect/runner/_observers.py", line 157, in __aenter__
2026-04-06 00:21:59.931 IST
    self._events_subscriber = await self._exit_stack.enter_async_context(
2026-04-06 00:21:59.931 IST
                              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2026-04-06 00:21:59.931 IST
    ...<5 lines>...
2026-04-06 00:21:59.931 IST
    )
2026-04-06 00:21:59.931 IST
    ^
2026-04-06 00:21:59.931 IST
  File "/usr/local/lib/python3.13/contextlib.py", line 668, in enter_async_context
2026-04-06 00:21:59.931 IST
    result = await _enter(cm)
2026-04-06 00:21:59.931 IST
             ^^^^^^^^^^^^^^^^
2026-04-06 00:21:59.931 IST
  File "/opt/venv/lib/python3.13/site-packages/prefect/events/clients.py", line 555, in __aenter__
2026-04-06 00:21:59.931 IST
    await self._reconnect()
2026-04-06 00:21:59.931 IST
  File "/opt/venv/lib/python3.13/site-packages/prefect/events/clients.py", line 566, in _reconnect
2026-04-06 00:21:59.931 IST
    self._websocket = await self._connect.__aenter__()
2026-04-06 00:21:59.931 IST
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2026-04-06 00:21:59.931 IST
  File "/opt/venv/lib/python3.13/site-packages/websockets/asyncio/client.py", line 587, in __aenter__
2026-04-06 00:21:59.931 IST
    return await self
2026-04-06 00:21:59.931 IST
           ^^^^^^^^^^
2026-04-06 00:21:59.931 IST
  File "/opt/venv/lib/python3.13/site-packages/websockets/asyncio/client.py", line 578, in __await_impl__
2026-04-06 00:21:59.931 IST
    raise TimeoutError("timed out during opening handshake") from exc
2026-04-06 00:21:59.932 IST
An exception occurred.
2026-04-06 00:22:00.475 IST

Container called exit(1).
2026-04-06 00:22:00.475 IST
TimeoutError: timed out during opening handshake
2026-04-06 00:22:06.079 IST


Cloud Run

RunJob

prefect-worker-prod-c0b28f7-qvj9w
Execution prefect-worker-prod-c0b28f7-qvj9w has failed to complete, 0/1 tasks were a success.
m
Your message was too long, here's your message at the allowed limit: I am facing the following error around websocket timeouts suggest a suitable way to reliably prevent this or work around it to prevent flow run crashes ``` 49 results Severity Time Summary Loading Scanned up to 4/6/26 7 11 AM 2026-04-06 00 21 59 929 IST Traceback (most recent call last) File /opt/venv/lib/python3 13/site-packages/websockets/asyncio/client py line 543 in __await_impl__ await self connection handshake( 2026-04-06 00 21 59 929 IST <2 lines> 2026-04-06 00 21 59 929 IST ) 2026-04-06 00 21 59 929 IST File /opt/venv/lib/python3 13/site-packages/websockets/asyncio/client py line 104 in handshake 2026-04-06 00 21 59 929 IST await asyncio wait( 2026-04-06 00 21 59 929 IST <2 lines> 2026-04-06 00 21 59 929 IST ) 2026-04-06 00 21 59 929 IST File /usr/local/lib/python3 13/asyncio/tasks py line 451 in wait 2026-04-06 00 21 59 929 IST return await _wait(fs timeout return_when loop) 2026-04-06 00 21 59 929 IST ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ 2026-04-06 00 21 59 929 IST File /usr/local/lib/python3 13/asyncio/tasks py line 537 in _wait 2026-04-06 00 21 59 929 IST await waiter 2026-04-06 00 21 59 929 IST asyncio exceptions CancelledError 2026-04-06 00 21 59 929 IST The above exception was the direct cause of the following exception 2026-04-06 00 21 59 929 IST Traceback (most recent call last) File /opt/venv/lib/python3 13/site-packages/websockets/asyncio/client py line 539 in __await_impl__ async with asyncio_timeout(self open_timeout) ~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^ File /usr/local/lib/python3 13/asyncio/timeouts py line 116 in aexit raise TimeoutError from exc_val 2026-04-06 00 21 59 929 IST TimeoutError 2026-04-06 00 21 59 929 IST The above exception was the direct cause of the following exception 2026-04-06 00 21 59 931 IST Traceback (most recent call last) File /opt/venv/lib/python3 13/site-packages/prefect/cli/_utilities py line 44 in wrapper return fn(*args *kwargs) File /opt/venv/lib/python3 13/site-packages/prefect/cli/_types py line 159 in sync_fn return asyncio run(async_fn(*args *kwargs)) ~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^ File /usr/local/lib/python3 13/asyncio/runners py line 195 in run return runner run(main) ~~~~~~~~~~^^^^^^ File /usr/local/lib/python3 13/asyncio/runners py line 118 in run return self _loop run_until_complete(task) ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^ File /usr/local/lib/python3 13/asyncio/base_events py line 725 in run_until_complete return future result() ~~~~~~~~~~~~~^^ File /opt/venv/lib/python3 13/site-packages/prefect/cli/flow_run py line 412 in execute await runner execute_flow_run(id) File /opt/venv/lib/python3 13/site-packages/prefect/runner/runner py line 601 in execute_flow_run async with context ^^^^^^^ File /opt/venv/lib/python3 13/site-packages/prefect/runner/runner py line 1571 in aenter self _cancelling_observer = await self _exit_stack enter_async_context( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ 2026-04-06 00 21 59 931 IST <6 lines> 2026-04-06 00 21 59 931 IST ) 2026-04-06 00 21 59 931 IST ^ 2026-04-06 00 21 59 931 IST File /usr/local/lib/python3 13/contextlib py line 668 in enter_async_context 2026-04-06 00 21 59 931 IST result = await _enter(cm) 2026-04-06 00 21 59 931 IST ^^^^^^^^^^^^^^^^ 2026-04-06 00 21 59 931 IST File /opt/venv/lib/python3 13/site-packages/prefect/runner/_observers py line 157 in aenter 2026-04-06 00 21 59 931 IST self _events_subscriber = await self _exit_stack enter_async_context( 2026-04-06 00 21 59 931 IST ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ 2026-04-06 00 21 59 931 IST <5 lines> 2026-04-06 00 21 59 931 IST ) 2026-04-06 00 21 59 931 IST ^ 2026-04-06 00 21 59 931 IST File /usr/local/lib/python3 13/contextlib py line 668 in enter_async_context 2026-04-06 00 21 59 931 IST result = await
g
ok @Marvin based on this, suggest what possible options are available to reliably prevent these failures
m
✅ thought for 394.4 seconds
Got it — assuming you’re on Prefect 3.x, here are reliable, “battle-tested” options you can combine to prevent or absorb transient failures. I grouped them by failure pattern and included minimal examples and docs links. 1) Transient network/API errors (timeouts, 5xx/429s) - Retries with exponential backoff + jitter (prevents thundering herds) - Optional conditional retries to skip unrecoverable errors - Add explicit timeouts so stuck calls fail fast and retry Links: - Retries, backoff, jitter Example:
Copy code
from prefect import task, flow
from prefect.tasks import exponential_backoff
import httpx

def retry_condition(task, task_run, state) -> bool:
    try:
        state.result()
    except httpx.HTTPStatusError as e:
        return e.response.status_code in (500, 502, 503, 504, 429)
    except Exception:
        return True  # retry all other transient errors

@task(
    retries=5,
    retry_delay_seconds=exponential_backoff(2),
    retry_jitter_factor=2,
    retry_condition_fn=retry_condition,
    timeout_seconds=30,
)
def call_api(url: str):
    r = httpx.get(url, timeout=20)
    r.raise_for_status()
    return r.json()

@flow
def pipeline():
    return call_api("<https://service/api>")
2) API rate limits and shared-resource contention - Tag-based task concurrency limits for simple “N at a time” by tag - Global Concurrency Limits (GCLs) for fine-grained control and true rate limiting via slot decay Links: - Tag-based concurrency limits - Global concurrency limits Tag limit example:
Copy code
from prefect import task

@task(tags=["database"])
def query_db(...):
    ...
CLI (verified):
Copy code
prefect concurrency-limit create database 10
prefect concurrency-limit inspect database
Global Concurrency + rate limiting:
Copy code
from prefect import task
from prefect.concurrency.sync import concurrency, rate_limit

@task
def write_once():
    with concurrency("db-writes", occupy=1, strict=True):
        ...

@task
def call_rate_limited_api():
    rate_limit("ext-api")  # requires slot decay limit configured
    ...
CLI (verified):
Copy code
prefect gcl create db-writes --limit 1
prefect gcl create ext-api --limit 10 --slot-decay-per-second 1.0
3) Long or flaky tasks - Timeouts to bound execution and trigger retries - Persist results to prevent rework after restarts - Cache results to skip repeating expensive work Links: - Caching and cache policies Examples:
Copy code
from datetime import timedelta
from prefect import task
from prefect.cache_policies import INPUTS

@task(timeout_seconds=300, persist_result=True, cache_policy=INPUTS, cache_expiration=timedelta(hours=1))
def expensive_compute(x):
    ...
    return result
4) Prevent duplicate work and race conditions on identical inputs - Use cache policies keyed by inputs and raise the isolation level to SERIALIZABLE with a lock manager (single-writer behavior for a given key) - Useful when concurrent runs with same inputs can clash Link: - Customize caching behavior Sketch:
Copy code
from prefect import task
from prefect.cache_policies import INPUTS
from prefect.transactions import IsolationLevel
from prefect.locking.memory import MemoryLockManager  # or a distributed lock manager

cache_policy = INPUTS.configure(
    isolation_level=IsolationLevel.SERIALIZABLE,
    lock_manager=MemoryLockManager(),
)

@task(cache_policy=cache_policy)
def build_once(x):
    ...
5) Control parallelism for stability - Limit flow concurrency at the work-pool level - Limit in-flow parallelism with task runners Links: - Manage work pools
Work pool concurrency (CLI, verified):
Copy code
prefect work-pool set-concurrency-limit "my-pool" 5
Task runner concurrency:
Copy code
from prefect import flow
from prefect.task_runners import ThreadPoolTaskRunner

@flow(task_runner=ThreadPoolTaskRunner(max_workers=4))
def my_flow():
    ...
6) Early detection and auto-action - State hooks to notify/measure before/after retries, on failure Link: State change hooks Example:
Copy code
from prefect import task

def on_fail(task, task_run, state):
    print(f"{task_run.name} failed: {state.message}")

@task(on_failure=[on_fail])
def fragile():
    ...
7) Manual recovery safety net - Quickly reschedule failed runs after fixing upstreams Link: Retry flow runs CLI (verified):
Copy code
prefect flow-run retry <id-or-name>
Suggested recipes by symptom - HTTP 429s/5xx: retries + exponential backoff + jitter + timeouts + GCL rate_limit with slot_decay_per_second - DB lock/contention: tag-based concurrency OR GCL concurrency around critical sections - Duplicate workload clashes: cache_policy=INPUTS with SERIALIZABLE isolation + lock manager - Worker restarts during long tasks: persist_result=True and chunk work into smaller retriable steps - Intermittent library timeouts: shorter task timeouts and more—but bounded—retries with backoff If you can share a failing run example (error message/traceback), where it runs (worker type/pool), and whether the failures are rate limits, timeouts, crashes, or contention, I can tailor an exact config/code snippet for your case.
g
@Marvin but the issue here is regarding websocket handshake timeouts, answer more specifically around how to mitigate this issue
m
✅ thought for 135.8 seconds
Thanks — focusing specifically on WebSocket handshake timeouts in Prefect 3.x, here’s how to mitigate them reliably. Quick wins (set these on your workers/flows) - Increase the client connect/read timeout (covers the initial WS upgrade handshake)
Copy code
export PREFECT_API_REQUEST_TIMEOUT=120  # try 120–180s
- Make the client more resilient to transient network issues
Copy code
export PREFECT_CLIENT_MAX_RETRIES=10
  export PREFECT_CLIENT_RETRY_JITTER_FACTOR=0.5
  # optional: add more retryable status codes if your network/proxy returns them
  export PREFECT_CLIENT_RETRY_EXTRA_CODES=408,429,502,503
- Toggle HTTP/2 on/off (some proxies/LBs behave better one way or the other)
Copy code
export PREFECT_API_ENABLE_HTTP2=true   # or false; test both
If you’re behind a proxy or strict firewall - Let the client use your proxy for the WebSocket upgrade
Copy code
export HTTPS_PROXY=<http://proxy.company.com:8443>
  export HTTP_PROXY=<http://proxy.company.com:8080>
  export NO_PROXY=localhost,127.0.0.1,<your-internal-hosts>
- Ensure the proxy/LB allows WebSocket Upgrade/Connection headers and long-lived connections on port 443. - Provide your corporate CA so TLS validation succeeds during the handshake
Copy code
export PREFECT_API_SSL_CERT_FILE=/path/to/ca-bundle.crt   # or SSL_CERT_FILE
Avoid disabling verification in production; if you must test quickly:
Copy code
export PREFECT_API_TLS_INSECURE_SKIP_VERIFY=true
Validate the API URL and DNS - Confirm 
PREFECT_API_URL
is correct and reachable from the runtime environment. If self-hosting behind a proxy/LB, prefer the closest/internal URL and add it to 
NO_PROXY
to skip the proxy if appropriate. Reduce handshake “bursts” - If many runs start at once, their WS handshakes can collide at the proxy/LB. Stagger starts slightly (e.g., jitter your schedules/automations) or lower worker concurrency temporarily to smooth spikes. Kubernetes-specific notes - Add the env vars above to your worker/job/pod templates. - Ensure cluster egress to 443 is permitted and not rewriting/stripping Upgrade headers. - If using a cluster proxy, set HTTP(S)_PROXY and NO_PROXY for in-cluster services properly. What to expect and how to debug - Handshake timeouts typically affect event/log streaming; runs generally still execute. You may see warnings about failing to connect to the events/logs stream. - Turn on debug logging to capture handshake details:
Copy code
export PREFECT_DEBUG_MODE=1
- If you use Prefect’s events client directly, you can increase reconnection attempts and keepalive frequency in code (useful in custom emitters):
Copy code
from prefect.events.clients import get_events_client

  client = get_events_client(
      reconnection_attempts=15,  # default ~10
      # keepalives for long-lived connections (post-handshake)
      # checkpoint_every=700,
      # checkpoint_interval=30.0,
  )
Recommended baseline for flaky networks
Copy code
export PREFECT_API_REQUEST_TIMEOUT=180
export PREFECT_CLIENT_MAX_RETRIES=12
export PREFECT_CLIENT_RETRY_JITTER_FACTOR=0.4
# try both to see which your proxy/LB prefers
export PREFECT_API_ENABLE_HTTP2=true
Docs to reference - Configuration and settings - Events (WebSocket-based) If you can share: - Where this occurs (workers vs. flows), Cloud vs. self-hosted - Whether you’re behind a proxy/LB (and if so, which) - A short log snippet with PREFECT_DEBUG_MODE=1 …I can recommend an exact set of env vars and any proxy/LB header tweaks to resolve your specific handshake timeouts.
g
@Marvin is this the appropriate way to set the job variable for a flow run ?
m
✅ thought for 71.4 seconds
Short answer: Yes — for a single flow run, pass job variables either: - Programmatically via 
run_deployment(..., job_variables={...})
- Or via CLI with 
--job-variable/-jv key=value
(value interpreted as JSON) Examples Python (verified by 
prefect.deployments.run_deployment
signature)
Copy code
from prefect.deployments import run_deployment

run = run_deployment(
    name="my-flow/my-deployment",
    parameters={"x": 1},
    job_variables={
        # common across infra types
        "env": {"FOO": "BAR"},
        # infra-specific (e.g., Docker/Kubernetes workers)
        "image": "myrepo/app:1.2.3",
        # k8s example keys: namespace, labels, annotations, resources, etc.
        # process worker example: stream_output, env
    },
)
CLI (verified)
Copy code
# Multiple -jv flags; values must be JSON
prefect deployment run "my-flow/my-deployment" \
  -jv env='{"FOO":"BAR"}' \
  -jv image='"myrepo/app:1.2.3"'
Notes and gotchas - Job variables override the deployment’s default job settings for that one run; they do not change flow parameters. - The keys you can set depend on your worker type: - Process worker: env, stream_output, etc. - Docker worker: image, env, command, volumes, network, etc. - Kubernetes worker: image, env, namespace, resources, labels/annotations, pod spec fields, etc. - In the CLI, values are parsed as JSON. For strings, wrap the value in JSON quotes (e.g., 
image='"my:tag"'
). For objects, provide valid JSON (e.g., 
env='{"FOO":"BAR"}'
). If you share the exact command or snippet you’re using, I can sanity-check the quoting and keys for your worker type.
g
@Marvin { "env": { "PREFECT_DEBUG_MODE": 1 } } does this format work ?
m
✅ thought for 28.3 seconds
Short answer: Yes, that shape is correct for 
job_variables.env
. For reliability, make the value a string since OS env vars are strings. - Preferred JSON for a one-off run:
Copy code
{
  "env": {
    "PREFECT_DEBUG_MODE": "1"
  }
}
- Python:
Copy code
run_deployment(
  name="my-flow/my-deployment",
  job_variables={"env": {"PREFECT_DEBUG_MODE": "1"}}
)
- CLI (values parsed as JSON):
Copy code
prefect deployment run "my-flow/my-deployment" \
  -jv env='{"PREFECT_DEBUG_MODE":"1"}'
Notes: - Prefect parses "1"/"true" as truthy for `PREFECT_DEBUG_MODE`; using a string is safest. - This overrides the env just for that flow run; it won’t change your deployment defaults. - Debug mode is verbose—use sparingly in production.
g
Copy code
06:52:03.396 | DEBUG   | prefect.events.clients - EventsClient(id=140478236101520): Sending event id=019d6190-0b50-78ae-a8c8-e3bf2561d61f.

2026-04-06 12:22:03.399 IST
06:52:03.398 | DEBUG   | prefect.events.clients - EventsClient(id=140478236101520): Checkpointing event id=019d6190-0b50-78ae-a8c8-e3bf2561d61f.
2026-04-06 12:22:03.406 IST
06:52:03.404 | DEBUG   | EventsWorkerThread | prefect._internal.concurrency - Running call get() in thread 'EventsWorkerThread'
2026-04-06 12:22:03.406 IST
06:52:03.404 | DEBUG   | EventsWorkerThread | prefect._internal.concurrency - <WatcherThreadCancelScope, name='get' RUNNING, runtime=0.00> entered
2026-04-06 12:22:03.406 IST
06:52:03.405 | DEBUG   | EventsWorkerThread | prefect._internal.concurrency - <WatcherThreadCancelScope, name='get' COMPLETED, runtime=0.00> exited
2026-04-06 12:22:03.407 IST
06:52:03.405 | DEBUG   | EventsWorkerThread | prefect._internal.concurrency - Finished call get(<dropped>)
2026-04-06 12:22:03.407 IST
06:52:03.405 | DEBUG   | GlobalEventLoopThread | prefect._internal.concurrency - Service <prefect.events.worker.EventsWorker object at 0x7fc3a35dcad0> handling item Event(occurred=datetime.datetime(2026, 4, 6, 6, 52, 3, 297939, tzinfo=zoneinfo.ZoneInfo(key='UTC')), event='prefect.task-run.Completed', resource=Resource(root={'prefect.resource.id': 'prefect.task-run.019d6190-0b47-764f-8b17-ede317c5e037', 'prefect.resource.name': 'sync_skill_risk_metrics-ff9', 'prefect.run-count': '1', 'prefect.state-message': '', 'prefect.state-name': 'Completed', 'prefect.state-timestamp': '2026-04-06T06:52:03.297939+00:00', 'prefect.state-type': 'COMPLETED', 'prefect.orchestration': 'client'}), related=[], payload={'intended': {'from': 'RUNNING', 'to': 'COMPLETED'}, 'initial_state': {'type': 'RUNNING', 'name': 'Running', 'message': '', 'state_details': {'traceparent': '00-8ba1c24817a6928f59246cf5db4a491b-fd4f606742c9934b-01'}}, 'validated_state': {'type': 'COMPLETED', 'name': 'Completed', 'message': '', 'state_details': {'untrackable_result': True, 'traceparent': '00-8ba1c24817a6928f59246cf5db4a491b-fd4f606742c9934b-01'}, 'data': None}, 'task_run': {'name': 'sync_skill_risk_metrics-ff9', 'task_key': 'sync_skill_risk_metrics-2f5de74e', 'dynamic_key': 'ff9403fd-82de-4a63-bead-4fe539f7ad33', 'empirical_policy': {'max_retries': 0, 'retry_delay_seconds': 0.0, 'retries': 0, 'retry_delay': 0}, 'tags': [], 'labels': {'__OTEL_TRACEPARENT': '00-8ba1c24817a6928f59246cf5db4a491b-fd4f606742c9934b-01'}, 'task_inputs': {'scanner': []}, 'run_count': 1, 'flow_run_run_count': 1, 'expected_start_time': '2026-04-06T06:52:03.271755Z', 'start_time': '2026-04-06T06:52:03.280375Z', 'end_time': '2026-04-06T06:52:03.297939Z', 'total_run_time': 0.017564}}, id=UUID('019d6190-0b61-7fbc-8108-de48c7a18064'), follows=UUID('019d6190-0b50-78ae-a8c8-e3bf2561d61f'))
2026-04-06 12:22:03.407 IST
06:52:03.406 | DEBUG   | prefect.events.clients - EventsClient(id=140478236101520): Emitting event id=019d6190-0b61-7fbc-8108-de48c7a18064.
2026-04-06 12:22:03.407 IST
06:52:03.406 | DEBUG   | prefect.events.clients - Added event id=019d6190-0b61-7fbc-8108-de48c7a18064 to unconfirmed events list. There are now 3 unconfirmed events.
2026-04-06 12:22:03.408 IST
06:52:03.406 | DEBUG   | prefect.events.clients - EventsClient(id=140478236101520): Emit reconnection attempt 0.
2026-04-06 12:22:03.408 IST
06:52:03.407 | DEBUG   | prefect.events.clients - EventsClient(id=140478236101520): Sending event id=019d6190-0b61-7fbc-8108-de48c7a18064.
2026-04-06 12:22:03.409 IST
06:52:03.407 | DEBUG   | prefect.events.clients - EventsClient(id=140478236101520): Checkpointing event id=019d6190-0b61-7fbc-8108-de48c7a18064.
2026-04-06 12:22:03.409 IST
06:52:03.408 | DEBUG   | EventsWorkerThread | prefect._internal.concurrency - Running call get() in thread 'EventsWorkerThread'
2026-04-06 12:22:03.409 IST
06:52:03.408 | DEBUG   | EventsWorkerThread | prefect._internal.concurrency - <WatcherThreadCancelScope, name='get' RUNNING, runtime=0.00> entered
2026-04-06 12:22:03.757 IST
06:52:03.756 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - <WatcherThreadCancelScope, name='get' COMPLETED, runtime=0.44> exited
2026-04-06 12:22:03.758 IST
06:52:03.756 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - Encountered exception in call get(<dropped>)
2026-04-06 12:22:03.758 IST

Traceback (most recent call last):   File "/opt/venv/lib/python3.13/site-packages/prefect/_internal/concurrency/calls.py", line 399, in _run_sync     result = self.fn(*self.args, **self.kwargs)   File "/usr/local/lib/python3.13/queue.py", line 209, in get     raise Empty
2026-04-06 12:22:03.758 IST
_queue.Empty
2026-04-06 12:22:03.758 IST
06:52:03.757 | DEBUG   | GlobalEventLoopThread | prefect._internal.concurrency - Service <prefect.logging.handlers.APILogWorker object at 0x7fc3c89fcc20> processing batch of size 1949
2026-04-06 12:22:03.783 IST
06:52:03.782 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - Running call get(timeout=9.999976005999997) in thread 'APILogWorkerThread'
2026-04-06 12:22:03.783 IST
06:52:03.782 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - <WatcherThreadCancelScope, name='get' RUNNING, runtime=0.00> entered
2026-04-06 12:22:07.343 IST
06:52:07.342 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - <WatcherThreadCancelScope, name='get' COMPLETED, runtime=10.00> exited
2026-04-06 12:22:07.343 IST
06:52:07.342 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - Encountered exception in call get(<dropped>)
2026-04-06 12:22:07.343 IST

Traceback (most recent call last):   File "/opt/venv/lib/python3.13/site-packages/prefect/_internal/concurrency/calls.py", line 399, in _run_sync     result = self.fn(*self.args, **self.kwargs)   File "/usr/local/lib/python3.13/queue.py", line 209, in get     raise Empty
2026-04-06 12:22:07.343 IST
_queue.Empty
2026-04-06 12:22:07.344 IST
06:52:07.343 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - Running call get(timeout=9.999977265000041) in thread 'APILogWorkerThread'
2026-04-06 12:22:07.344 IST
06:52:07.343 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - <WatcherThreadCancelScope, name='get' RUNNING, runtime=0.00> entered
2026-04-06 12:22:13.784 IST
06:52:13.782 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - <WatcherThreadCancelScope, name='get' COMPLETED, runtime=10.00> exited
2026-04-06 12:22:13.785 IST
06:52:13.782 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - Encountered exception in call get(<dropped>)
2026-04-06 12:22:13.785 IST

Traceback (most recent call last):   File "/opt/venv/lib/python3.13/site-packages/prefect/_internal/concurrency/calls.py", line 399, in _run_sync     result = self.fn(*self.args, **self.kwargs)   File "/usr/local/lib/python3.13/queue.py", line 209, in get     raise Empty
2026-04-06 12:22:13.785 IST
_queue.Empty
2026-04-06 12:22:13.785 IST
06:52:13.783 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - Running call get(timeout=9.999956994999991) in thread 'APILogWorkerThread'
2026-04-06 12:22:13.785 IST
06:52:13.783 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - <WatcherThreadCancelScope, name='get' RUNNING, runtime=0.00> entered
2026-04-06 12:22:17.345 IST
06:52:17.343 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - <WatcherThreadCancelScope, name='get' COMPLETED, runtime=10.00> exited
2026-04-06 12:22:17.346 IST
06:52:17.343 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - Encountered exception in call get(<dropped>)
2026-04-06 12:22:17.346 IST

Traceback (most recent call last):   File "/opt/venv/lib/python3.13/site-packages/prefect/_internal/concurrency/calls.py", line 399, in _run_sync     result = self.fn(*self.args, **self.kwargs)   File "/usr/local/lib/python3.13/queue.py", line 209, in get     raise Empty
2026-04-06 12:22:17.346 IST
_queue.Empty
2026-04-06 12:22:17.346 IST
06:52:17.344 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - Running call get(timeout=9.999975950000021) in thread 'APILogWorkerThread'
2026-04-06 12:22:17.346 IST
06:52:17.345 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - <WatcherThreadCancelScope, name='get' RUNNING, runtime=0.00> entered
2026-04-06 12:22:23.785 IST
06:52:23.784 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - <WatcherThreadCancelScope, name='get' COMPLETED, runtime=10.00> exited
2026-04-06 12:22:23.785 IST
06:52:23.784 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - Encountered exception in call get(<dropped>)
2026-04-06 12:22:23.785 IST

Traceback (most recent call last):   File "/opt/venv/lib/python3.13/site-packages/prefect/_internal/concurrency/calls.py", line 399, in _run_sync     result = self.fn(*self.args, **self.kwargs)   File "/usr/local/lib/python3.13/queue.py", line 209, in get     raise Empty
2026-04-06 12:22:23.785 IST
_queue.Empty
2026-04-06 12:22:23.786 IST
06:52:23.785 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - Running call get(timeout=9.999975341000038) in thread 'APILogWorkerThread'
2026-04-06 12:22:23.786 IST
06:52:23.785 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - <WatcherThreadCancelScope, name='get' RUNNING, runtime=0.00> entered
2026-04-06 12:22:27.347 IST
06:52:27.345 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - <WatcherThreadCancelScope, name='get' COMPLETED, runtime=10.00> exited
2026-04-06 12:22:27.347 IST
06:52:27.345 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - Encountered exception in call get(<dropped>)
2026-04-06 12:22:27.347 IST

Traceback (most recent call last):   File "/opt/venv/lib/python3.13/site-
@Marvin check the logs above from debug run and pinpoint how to prevent frequent timeouts / handshake failures
m
Your message was too long, here's your message at the allowed limit: ```06 52 03 396 | DEBUG | prefect events clients - EventsClient(id=140478236101520) Sending event id=019d6190-0b50-78ae-a8c8-e3bf2561d61f 2026-04-06 12 22 03 399 IST 06 52 03 398 | DEBUG | prefect events clients - EventsClient(id=140478236101520) Checkpointing event id=019d6190-0b50-78ae-a8c8-e3bf2561d61f 2026-04-06 12 22 03 406 IST 06 52 03 404 | DEBUG | EventsWorkerThread | prefect _internal concurrency - Running call get() in thread 'EventsWorkerThread' 2026-04-06 12 22 03 406 IST 06 52 03 404 | DEBUG | EventsWorkerThread | prefect _internal concurrency - <WatcherThreadCancelScope name='get' RUNNING runtime=0 00> entered 2026-04-06 12 22 03 406 IST 06 52 03 405 | DEBUG | EventsWorkerThread | prefect _internal concurrency - <WatcherThreadCancelScope name='get' COMPLETED runtime=0 00> exited 2026-04-06 12 22 03 407 IST 06 52 03 405 | DEBUG | EventsWorkerThread | prefect _internal concurrency - Finished call get(<dropped>) 2026-04-06 12 22 03 407 IST 06 52 03 405 | DEBUG | GlobalEventLoopThread | prefect _internal concurrency - Service <prefect events worker EventsWorker object at 0x7fc3a35dcad0> handling item Event(occurred=datetime datetime(2026 4 6 6 52 3 297939 tzinfo=zoneinfo ZoneInfo(key='UTC')) event='prefect task-run Completed' resource=Resource(root={'prefect resource id' 'prefect task-run 019d6190-0b47-764f-8b17-ede317c5e037' 'prefect resource name' 'sync_skill_risk_metrics-ff9' 'prefect run-count' '1' 'prefect state-message' '' 'prefect state-name' 'Completed' 'prefect state-timestamp' '2026-04-06T06 52 03 297939+00 00' 'prefect state-type' 'COMPLETED' 'prefect orchestration' 'client'}) related=[] payload={'intended' {'from' 'RUNNING' 'to' 'COMPLETED'} 'initial_state' {'type' 'RUNNING' 'name' 'Running' 'message' '' 'state_details' {'traceparent' '00-8ba1c24817a6928f59246cf5db4a491b-fd4f606742c9934b-01'}} 'validated_state' {'type' 'COMPLETED' 'name' 'Completed' 'message' '' 'state_details' {'untrackable_result' True 'traceparent' '00-8ba1c24817a6928f59246cf5db4a491b-fd4f606742c9934b-01'} 'data' None} 'task_run' {'name' 'sync_skill_risk_metrics-ff9' 'task_key' 'sync_skill_risk_metrics-2f5de74e' 'dynamic_key' 'ff9403fd-82de-4a63-bead-4fe539f7ad33' 'empirical_policy' {'max_retries' 0 'retry_delay_seconds' 0 0 'retries' 0 'retry_delay' 0} 'tags' [] 'labels' {'__OTEL_TRACEPARENT' '00-8ba1c24817a6928f59246cf5db4a491b-fd4f606742c9934b-01'} 'task_inputs' {'scanner' []} 'run_count' 1 'flow_run_run_count' 1 'expected_start_time' '2026-04-06T06 52 03 271755Z' 'start_time' '2026-04-06T06 52 03 280375Z' 'end_time' '2026-04-06T06 52 03 297939Z' 'total_run_time' 0 017564}} id=UUID('019d6190-0b61-7fbc-8108-de48c7a18064') follows=UUID('019d6190-0b50-78ae-a8c8-e3bf2561d61f')) 2026-04-06 12 22 03 407 IST 06 52 03 406 | DEBUG | prefect events clients - EventsClient(id=140478236101520) Emitting event id=019d6190-0b61-7fbc-8108-de48c7a18064 2026-04-06 12 22 03 407 IST 06 52 03 406 | DEBUG | prefect events clients - Added event id=019d6190-0b61-7fbc-8108-de48c7a18064 to unconfirmed events list There are now 3 unconfirmed events 2026-04-06 12 22 03 408 IST 06 52 03 406 | DEBUG | prefect events clients - EventsClient(id=140478236101520) Emit reconnection attempt 0 2026-04-06 12 22 03 408 IST 06 52 03 407 | DEBUG | prefect events clients - EventsClient(id=140478236101520) Sending event id=019d6190-0b61-7fbc-8108-de48c7a18064 2026-04-06 12 22 03 409 IST 06 52 03 407 | DEBUG | prefect events clients - EventsClient(id=140478236101520) Checkpointing event id=019d6190-0b61-7fbc-8108-de48c7a18064 2026-04-06 12 22 03 409 IST 06 52 03 408 | DEBUG | EventsWorkerThread | prefect _internal concurrency - Running call get() in thread 'EventsWorkerThread' 2026-04-06 12 22 03 409 IST 06 52 03 408 | DEBUG | EventsWorkerThread | prefect _internal concurrency - <WatcherThreadCancelScope name='get' RUNNING runtime=0 00> entered 2026-04-06 12 22 03 757 IST 06 52 03
756 | DEBUG | APILogWorkerThread | prefect _internal concurrency - <WatcherThreadCancelScope name='get' COMPLETED runtime=0 44> exited 2026-04-06 12 22 03 758 IST 06 52 03 756 | DEBUG | APILogWorkerThread | prefect _internal concurrency - Encountered exception in call get(<dropped>)
g
@Marvin continue with this and answer the question
m
✅ thought for 39.9 seconds
Yes — using 
job_variables.env
is the right way to set environment variables for a single flow run. For your specific example, use string values to avoid type issues with OS env vars. - Correct JSON payload for `job_variables`:
Copy code
{
  "env": {
    "PREFECT_DEBUG_MODE": "1"
  }
}
- Python:
Copy code
from prefect.deployments import run_deployment

run_deployment(
  name="my-flow/my-deployment",
  job_variables={
    "env": {
      "PREFECT_DEBUG_MODE": "1"
    }
  }
)
- CLI (values parsed as JSON):
Copy code
prefect deployment run "my-flow/my-deployment" \
  -jv env='{"PREFECT_DEBUG_MODE":"1"}'
Why strings? Even though some infra may coerce numbers to strings, environment variables are strings by convention. Using 
"1"
avoids surprises across worker types (process, docker, k8s). Tips and gotchas - Scope: 
job_variables.env
applies to the flow run process/container only. It will not affect the worker process. If you need debug logs for worker-level connectivity (e.g., handshake from the worker itself), set 
PREFECT_DEBUG_MODE=1
in the worker’s environment or work pool configuration as well. - Multiple variables: pass them in the same 
env
object:
Copy code
-jv env='{
  "PREFECT_DEBUG_MODE":"1",
  "PREFECT_API_REQUEST_TIMEOUT":"180",
  "PREFECT_CLIENT_MAX_RETRIES":"10"
}'
- Parameters vs job variables: 
parameters
are inputs to your flow function; 
job_variables
control the execution environment (image, env, resources, etc.) for that one run. - Validation: You can quickly verify by printing the env inside a test flow:
Copy code
from prefect import flow
import os

@flow
def check_env():
    print("DEBUG =", os.environ.get("PREFECT_DEBUG_MODE"))

# Trigger with job_variables.env as above and confirm "DEBUG = 1"
References we verified - Python API: 
prefect.deployments.run_deployment(..., job_variables=...)
- CLI: 
prefect deployment run ... --job-variable/-jv key=value
(value is interpreted as JSON) If you share your worker type (process, docker, or k8s) and how you’re triggering the run (Python, CLI, or UI), I can provide an exact example tailored to that setup.
7 Views
Open in Slack
PreviousNext
Powered by