Data, I/O, and Validation Errors

Shared runtime error model

When a task fails while converting data or moving inline task inputs and outputs, flyte-sdk represents both failures as user-originated runtime errors. RuntimeDataValidationError and InlineIOMaxBytesBreached inherit from RuntimeUserError, which fixes kind to "user". BaseRuntimeError stores the other transport fields—code, kind, and worker—and passes the root-cause message to RuntimeError, so str(exception) is the message supplied by the concrete error.

from errors import InlineIOMaxBytesBreached, RuntimeDataValidationError

validation_error = RuntimeDataValidationError("output", "cannot convert value", "my_task")
size_error = InlineIOMaxBytesBreached("inline output is too large")

assert validation_error.kind == "user"
assert size_error.code == "InlineIOMaxBytesBreached"
assert size_error.worker == "user"

The first constructor accepts a variable name, an exception or string describing the failure, and an optional task name. It does not pass a worker to RuntimeUserError, so its inherited worker value is None. The second accepts only a message and explicitly passes "user" as the worker. Runtime error conversion can carry these fields into execution-error protobufs; convert_error_to_native reconstructs native runtime exceptions from server errors.

Serialization and deserialization validation

A task that declares no outputs but returns a value fails during native-output conversion. The same error is used when the type engine cannot serialize a declared output. You normally encounter it by returning the wrong kind of value from a task rather than by constructing the exception yourself.

The runtime normalizes a non-tuple return value to a one-item tuple, then uses the task's NativeInterface to decide how to process it. For a no-output interface, None is accepted, but any other single value raises RuntimeDataValidationError with variable name "o0":

async def convert_from_native_to_outputs(o: Any, interface: NativeInterface, task_name: str = "") -> Outputs:
    # Always make it a tuple even if it's just one item to simplify logic below
    if not isinstance(o, tuple):
        o = (o,)

    if len(interface.outputs) == 0:
        if len(o) != 0:
            if len(o) == 1 and o[0] is not None:
                raise flyte.errors.RuntimeDataValidationError(
                    "o0",
                    f"Expected no outputs but got {o},did you miss a return type annotation?",
                    task_name,
                )

For declared outputs, _internal/runtime/convert.py walks the interface's (output_name, python_type) pairs and calls TypeEngine.to_literal for each returned value. Only TypeTransformerFailedError is wrapped at this point; the wrapper records the declared output name and task name:

async def convert_from_native_to_outputs(o: Any, interface: NativeInterface, task_name: str = "") -> Outputs:
    # Always make it a tuple even if it's just one item to simplify logic below
    if not isinstance(o, tuple):
        o = (o,)

    if len(interface.outputs) == 0:
        if len(o) != 0:
            if len(o) == 1 and o[0] is not None:
                raise flyte.errors.RuntimeDataValidationError(
                    "o0",
                    f"Expected no outputs but got {o},did you miss a return type annotation?",
                    task_name,
                )
    else:
        assert len(o) == len(interface.outputs), (
            f"Received {len(o)} outputs but return annotation has {len(interface.outputs)} outputs specified. "
        )
    named = []
    for (output_name, python_type), v in zip(interface.outputs.items(), o):
        try:
            lit = await TypeEngine.to_literal(v, python_type, TypeEngine.to_literal_type(python_type))
            named.append(run_definition_pb2.NamedLiteral(name=output_name, value=lit))
        except TypeTransformerFailedError as e:
            raise flyte.errors.RuntimeDataValidationError(output_name, e, task_name)

    return Outputs(proto_outputs=run_definition_pb2.Outputs(literals=named))

The complete function is convert_from_native_to_outputs in _internal/runtime/convert.py. The local controller calls it with the task's output, native interface, and name. The normal runtime path in _internal/runtime/taskrunner.py also calls it after the user function runs and before output upload:

return await convert_from_native_to_outputs(out, task.native_interface, task.name), None

A count mismatch is different

Do not assume every output-shape problem produces RuntimeDataValidationError. If the task declares outputs and the returned tuple has a different length, convert_from_native_to_outputs uses a bare assert:

assert len(o) == len(interface.outputs), (
    f"Received {len(o)} outputs but return annotation has {len(interface.outputs)} outputs specified. "
)

That path produces AssertionError, not the runtime validation error. The explicit RuntimeDataValidationError paths are the no-output/non-None case and a caught TypeTransformerFailedError from literal conversion. Other exceptions raised by conversion are not explicitly caught by this function.

The emitted validation code is spelled "DataValiationError" in errors.py. Preserve that spelling when matching the error code, even though the class name is RuntimeDataValidationError.

Inline input and output byte limits

If serialized inline task data is larger than the configured limit, flyte-sdk raises InlineIOMaxBytesBreached instead of proceeding with the relevant storage operation or parsing the downloaded protobuf. Configure the limit on the task decorator when you create the task:

import flyte

env = flyte.TaskEnvironment(name="my_env", image="my_image", resources=flyte.Resources(cpu="1", memory="1Gi"))

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

TaskEnvironment.task has max_inline_io_bytes: int = MAX_INLINE_IO_BYTES. It passes the value into the resulting AsyncFunctionTaskTemplate, where it is stored on the task template. The task documentation limits this setting to data passed directly to the task, such as primitives, strings, and dictionaries; it does not apply to files, directories, or dataframes.

For a particular invocation, use TaskTemplate.override:

small_inline_task = produce_value.override(max_inline_io_bytes=256 * 1024)

override creates a replacement template with the selected value. Its implementation uses max_inline_io_bytes = max_inline_io_bytes or self.max_inline_io_bytes, so passing 0 does not create a zero-byte limit; 0 falls back to the existing task value.

The default constant in models.py is:

MAX_INLINE_IO_BYTES = 10 * 1024 * 1024  # 100 MB

The numeric value is 10 MiB. The inline comment says 100 MB, while the task and remote-task comments describe 10 MB; use the numeric constant as the authoritative default. The remote task-reference model also defaults its field to 10 * 1024 * 1024.

Where the limit is checked

For an ordinary remote task call, the controller serializes the inputs first and passes the task's configured value to upload_inputs_with_retry:

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"
    )

This check occurs before storage.put_stream. Storage failures take a different path: the helper converts them to RuntimeSystemError. The same input-upload and realized-output-loading flow is used for remote task references, with the task-reference limit passed into the controller.

When remote outputs are realized, load_and_convert_outputs calls io.load_outputs(..., max_bytes=max_bytes) before converting protobuf literals back to native values. The lower-level I/O checks are:

• _internal/runtime/io.py:upload_outputs compares outputs.proto_outputs.ByteSize() with max_bytes before serializing and uploading the output protobuf.
• io.load_inputs reads storage chunks and raises when cumulative raw chunk length exceeds the limit.
• io.load_outputs applies the same cumulative chunk check and only calls ParseFromString after all accepted chunks have been collected.

The low-level functions use -1 as the unlimited sentinel:

async def load_outputs(path: str, max_bytes: int = -1) -> Outputs:
    """
    :param path: output file to be loaded
    :param max_bytes: Maximum number of bytes to read from the output file.
                      If -1, reads the entire file.
    :return: Outputs object
    """
    lm = run_definition_pb2.Outputs()

    if max_bytes == -1:
        proto_str = b"".join([c async for c in storage.get_stream(path=path)])
    else:
        proto_bytes = []
        total_bytes = 0
        async for chunk in storage.get_stream(path=path):
            if total_bytes + len(chunk) > max_bytes:
                import flyte.errors

                raise flyte.errors.InlineIOMaxBytesBreached(
                    f"Output file at {path} exceeds max_bytes limit of {max_bytes}"
                )
            proto_bytes.append(chunk)
            total_bytes += len(chunk)
        proto_str = b"".join(proto_bytes)

    lm.ParseFromString(proto_str)
    return Outputs(proto_outputs=lm)

Treat -1 as a low-level I/O convention, not as the normal task configuration example. The remote upload_inputs_with_retry helper compares len(serialized_inputs) > max_bytes directly and does not special-case -1; remote callers pass the configured task limit. Trace recording and replay separately use MAX_TRACE_BYTES, which is assigned from MAX_INLINE_IO_BYTES, rather than a task-specific limit.

Diagnosing the two errors

Error code	Trigger	Size or context reported	What to inspect
DataValiationError	A no-output task returns a non-None value, or TypeEngine.to_literal raises TypeTransformerFailedError for a declared output	Variable name and, when supplied, task name; the conversion exception is included	Return value and return annotation; the output type transformer
InlineIOMaxBytesBreached	Serialized remote inputs exceed the limit, an output protobuf is too large to upload, or streamed input/output data exceeds the limit	Remote input upload reports serialized byte length; output upload reports protobuf ByteSize(); downloads report the configured limit	max_inline_io_bytes, serialized data size, and whether the data is inline rather than a file, directory, or dataframe

The measurements are not uniform. Remote input upload measures the already serialized byte string with len; output upload measures the protobuf with ByteSize(); download paths add the raw lengths of received chunks. The download implementations collect accepted chunks in memory before parsing, so configuring a limit bounds what they accept but does not turn them into streaming protobuf parsers.

Finally, the error classes are user errors, but not every nearby failure is one: a remote storage exception is converted to RuntimeSystemError, and an output-count assertion is an AssertionError. These distinctions matter when you inspect code, kind, or the serialized execution error. The scoped flyte-sdk sources provide the production call sites for these behaviors; no matching repository test or Markdown example files were found in the explored source.