Handling Resource and Execution Errors

When a remote Flyte task fails because it ran out of memory, exceeded its runtime, consumed all retries, or produced an oversized inline payload, catch the typed exception and inspect both its message and structured runtime fields before changing the task configuration.

from flyte.errors import (
    InlineIOMaxBytesBreached,
    OOMError,
    RetriesExhaustedError,
    TaskTimeoutError,
)

try:
    result = remote_task()
except (OOMError, TaskTimeoutError, RetriesExhaustedError, InlineIOMaxBytesBreached) as exc:
    print(str(exc))       # root-cause message
    print(exc.code)        # normalized runtime error code
    print(exc.kind)        # "user"
    print(exc.worker)      # worker identifier, or None
    raise

All four exceptions are subclasses of RuntimeUserError, which calls BaseRuntimeError with kind="user". BaseRuntimeError passes the root-cause message to RuntimeError and stores code, kind, and worker. Consequently, the exception string is the message; inspect exc.code separately when classifying or logging the failure.

Diagnose the failure by symptom

Out of memory: OOMError

A server execution error is converted by _internal.runtime.convert.convert_error_to_native when its kind is USER and its original error code contains OOM, case-insensitively:

case execution_pb2.ExecutionError.USER:
    if "OOM" in err.code.upper():
        return flyte.errors.OOMError(code=user_code, message=err.message, worker=err.worker)

OOMError is a marker subclass and does not define a constructor of its own. It therefore accepts the inherited RuntimeUserError arguments: code, message, and optional worker. The remote controller obtains this converted exception through handle_action_failure and raises it for the failed action.

Use the worker and message to identify where the memory failure occurred. Then either reduce the task's memory use or adjust the task's resource configuration; TaskTemplate exposes a resources field alongside its retry and timeout settings. An OOM is not caused by the inline-I/O limit: that limit produces InlineIOMaxBytesBreached instead.

Runtime timeout: TaskTimeoutError

Configure the task's maximum runtime with Timeout and pass it to TaskEnvironment.task:

from datetime import timedelta

from flyte import TaskEnvironment, Timeout

env = TaskEnvironment(name="timeout_example")
timeout = Timeout(max_runtime=timedelta(minutes=5))

@env.task(timeout=timeout)
async def my_task():
    pass

Timeout also accepts max_queued_time:

from datetime import timedelta

from flyte import TaskEnvironment, Timeout
env = TaskEnvironment(name="timeout_example")

timeout = Timeout(
    max_runtime=timedelta(minutes=5),
    max_queued_time=timedelta(minutes=10),
)

@env.task(timeout=timeout)
async def my_task():
    pass

The timeout serializer currently emits only Timeout.max_runtime. get_proto_timeout calls timeout_from_request, converts an integer runtime to a timedelta when necessary, and writes the resulting runtime duration to the task protobuf. Although Timeout models a queue-time value, max_queued_time is not written by this serializer.

The remote controller detects timeout from the completed action phase, before its ordinary failed-action handling:

if n.phase == run_definition_pb2.PHASE_TIMED_OUT:
    raise flyte.errors.TaskTimeoutError(
        f"Action {n.action_id.name} timed out, raising exception in current Action {current_action_id.name}"
    )

if n.has_error() or n.phase == run_definition_pb2.PHASE_FAILED:
    exc = await handle_action_failure(action, _task.name)
    raise exc

TaskTimeoutError has a restricted constructor: it accepts only message and sets code to "TaskTimeoutError" and kind to "user". It is raised directly for PHASE_TIMED_OUT; it is not selected by convert_error_to_native.

TaskEnvironment.task also accepts an integer number of seconds or a timedelta through its timeout parameter. timeout_from_request rejects other types with ValueError. The decorator signature defaults timeout to 0, while TaskTemplate.timeout defaults to None; distinguish the decorator's default from an explicitly configured timeout when inspecting serialized task metadata.

All retries consumed: RetriesExhaustedError

Set a retry count directly or provide a RetryStrategy:

from flyte import RetryStrategy, TaskEnvironment

env = TaskEnvironment(name="retry_example")

@env.task(retries=5)
async def retrying_task():
    pass

@env.task(
    retries=RetryStrategy(count=5, backoff=10, backoff_factor=2),
)
async def retrying_task_with_strategy():
    pass

RetryStrategy defines count, backoff, and backoff_factor. During task serialization, get_proto_retry_strategy writes only retries.count:

def get_proto_retry_strategy(
    retries: RetryStrategy | int | None,
) -> Optional[literals_pb2.RetryStrategy]:
    if retries is None:
        return None

    if isinstance(retries, int):
        raise AssertionError("Retries should be an instance of RetryStrategy, not int")

    return literals_pb2.RetryStrategy(retries=retries.count)

The task APIs accept an integer and convert it to a RetryStrategy during TaskTemplate.__post_init__, so retries=5 is the convenient public form. The serialized task metadata carries the count, not the strategy's backoff values.

When a server USER error code contains the case-sensitive substring RetriesExhausted, convert_error_to_native constructs RetriesExhaustedError with the normalized code, message, and worker. Like OOMError, this is a marker subclass and uses the inherited (code, message, worker) constructor. Do not confuse task retries with remote-controller settings such as RPC or system retries; those are separate controller configuration values.

Oversized direct input or output: InlineIOMaxBytesBreached

Raise the per-task inline limit when defining the task:

from flyte import TaskEnvironment

env = TaskEnvironment(name="inline_io_example")

@env.task(max_inline_io_bytes=1024 * 1024)
async def bounded_task(value: str) -> str:
    return value

The task API defines this setting as the maximum size, in bytes, for direct inputs and outputs such as primitives, strings, and dictionaries. Its documentation explicitly excludes files, directories, and dataframes. The SDK-wide default is MAX_INLINE_IO_BYTES = 10 * 1024 * 1024 in models.py (10 MiB), although the source comment says 100 MB.

For remote submission, _internal.controllers.remote._controller.upload_inputs_with_retry checks the serialized input before calling storage:

if len(serialized_inputs) > max_bytes:
    raise flyte.errors.InlineIOMaxBytesBreached(
        f"Inputs exceed max_bytes limit of {max_bytes / 1024 / 1024} MB,"
        f" actual size: {len(serialized_inputs) / 1024 / 1024} MB"
    )
await storage.put_stream(serialized_inputs, to_path=inputs_uri)

The remote controller forwards the task's max_inline_io_bytes to this helper and uses the same value while loading task outputs:

return await load_and_convert_outputs(
    _task.native_interface, n.realized_outputs_uri, max_bytes=_task.max_inline_io_bytes
)

The runtime I/O layer applies the limit to serialized protobuf bytes. upload_outputs compares outputs.proto_outputs.ByteSize() with max_bytes; load_inputs and load_outputs count downloaded chunks and raise when adding the next chunk would exceed the limit. These loaders collect the accepted chunks and join them before parsing, so the check rejects an oversized serialized payload but is not a fully streaming load.

Move a large direct value to a file, directory, or dataframe when it should not be carried as inline task data. Increasing max_inline_io_bytes is the alternative when the serialized input or output is expected to remain direct task data.

What to inspect

Exception	Origin	code	str(exc)	worker
OOMError	convert_error_to_native for a USER code containing OOM	Cleaned server code	Server error message	Server-provided worker or None
TaskTimeoutError	Remote controller after PHASE_TIMED_OUT	TaskTimeoutError	Action/current-action timeout message	None unless assigned later; the constructor does not accept a worker
RetriesExhaustedError	convert_error_to_native for a USER code containing RetriesExhausted	Cleaned server code	Server error message	Server-provided worker or None
InlineIOMaxBytesBreached	Input upload, output upload, or input/output download	InlineIOMaxBytesBreached	Limit and measured-size message	None; the constructor does not accept a worker

The conversion path removes an optional server prefix separated by | when assigning the structured code, but OOM and retry matching uses the original server code. OOM matching is case-insensitive substring matching; RetriesExhausted matching is case-sensitive substring matching.

Configuration caveats

• The inline-I/O default is 10 * 1024 * 1024 bytes despite the models.py comment claiming 100 MB. The independently repeated remote-task default is also 10 * 1024 * 1024.
• Runtime I/O helpers treat max_bytes=-1 as unlimited in upload_outputs, load_inputs, and load_outputs. The remote input-upload helper compares directly with max_bytes and has no -1 special case, so the public remote-task default is the positive 10-MiB value rather than -1.
• RetryStrategy.backoff and backoff_factor are Python-level fields, but task wire serialization emits only count.
• Timeout.max_queued_time is represented by Timeout, but get_proto_timeout serializes only max_runtime.
• Despite its name, upload_inputs_with_retry currently has a TODO to add a retry decorator. Its shown implementation makes one storage.put_stream attempt and wraps an upload failure as RuntimeSystemError; it does not implement storage-upload retries itself.
• The four exception classes are not uniformly constructible. OOMError and RetriesExhaustedError inherit (code, message, worker); TaskTimeoutError and InlineIOMaxBytesBreached accept only a message. Application code should normally catch these exceptions rather than instantiate them.

The repository search for this topic found no matching test files, README files, or example files. The concrete snippets above come from the exception definitions, production remote-controller/runtime-I/O paths, and API docstrings in flyte-sdk.