perf(serializers): default to msgpack with cloudpickle fallback

py_src/taskito/__init__.py

@@ -44,6 +44,7 @@
     JsonSerializer,
     MsgPackSerializer,
     Serializer,
+    SmartSerializer,
 )
 from taskito.task import TaskWrapper
 from taskito.testing import MockResource, TestMode, TestResult, TestResults
@@ -83,6 +84,7 @@
     "SerializationError",
     "Serializer",
     "Signature",
+    "SmartSerializer",
     "SoftTimeoutError",
     "TaskCancelledError",
     "TaskFailedError",

py_src/taskito/app.py

@@ -52,7 +52,7 @@
 from taskito.proxies.built_in import register_builtin_handlers
 from taskito.proxies.metrics import ProxyMetrics
 from taskito.result import JobResult
-from taskito.serializers import CloudpickleSerializer, Serializer
+from taskito.serializers import Serializer, SmartSerializer
 from taskito.webhooks import WebhookManager
 
 if TYPE_CHECKING:
@@ -147,7 +147,8 @@ def __init__(
             default_priority: Default task priority (higher = more urgent).
             result_ttl: Auto-cleanup completed/dead jobs older than this many
                 seconds. None disables auto-cleanup.
-            serializer: Serializer for task payloads. Defaults to CloudpickleSerializer.
+            serializer: Serializer for task payloads. Defaults to SmartSerializer
+                (msgpack with cloudpickle fallback).
             middleware: List of global middleware instances applied to all tasks.
             backend: Storage backend — ``"sqlite"`` (default) or ``"postgres"``.
             db_url: PostgreSQL connection URL (required when backend is ``"postgres"``).
@@ -218,7 +219,7 @@ def __init__(
             "on_success": [],
             "on_failure": [],
         }
-        self._serializer: Serializer = serializer or CloudpickleSerializer()
+        self._serializer: Serializer = serializer or SmartSerializer()
         self._task_serializers: dict[str, Serializer] = {}
         self._task_idempotent: dict[str, bool] = {}
         self._task_compensates: dict[str, str] = {}
@@ -386,8 +387,15 @@ def close(self) -> None:
             self._batch_accumulator = None
 
     def _deserialize_payload(self, task_name: str, payload: bytes) -> tuple:
-        """Deserialize a job payload using the per-task or queue-level serializer."""
-        return self._get_serializer(task_name).loads(payload)  # type: ignore[no-any-return]
+        """Deserialize a job payload into an ``(args, kwargs)`` pair.
+
+        Serializers without a native tuple type (msgpack, JSON) flatten the
+        payload to ``[args, kwargs]`` with ``args`` as a list. The worker hands
+        ``args`` straight to PyO3, which requires a real tuple, so the shape is
+        normalized back to ``(tuple, dict)`` here regardless of serializer.
+        """
+        args, kwargs = self._get_serializer(task_name).loads(payload)
+        return tuple(args), kwargs
 
     def enqueue(
         self,

py_src/taskito/serializers.py

@@ -6,6 +6,49 @@
 from typing import Any, Protocol, runtime_checkable
 
 import cloudpickle
+import msgpack
+
+# Format tags for the envelope written by ``SmartSerializer``. A legacy
+# cloudpickle payload starts with the pickle protocol-2+ opcode ``\x80``, which
+# never collides with these tags — so untagged bytes are unambiguously legacy.
+_CODEC_CLOUDPICKLE = b"\x00"
+_CODEC_MSGPACK = b"\x01"
+
+# msgpack has no native tuple type and would silently flatten tuples to lists.
+# A custom ExtType preserves them so payloads round-trip with exact Python
+# semantics (a task returning ``(1, 2)`` gets back ``(1, 2)``, not ``[1, 2]``).
+_EXT_TUPLE = 0
+
+
+def _msgpack_default(obj: Any) -> Any:
+    """Encode types msgpack can't represent natively.
+
+    Exact tuples become a tagged ExtType (recursively packed). Tuple
+    *subclasses* (e.g. namedtuples) are not exact tuples — ``list(obj)`` would
+    drop their type — so they raise here and fall back to cloudpickle, which
+    preserves them. Anything else also raises and falls back.
+    """
+    if type(obj) is tuple:
+        return msgpack.ExtType(_EXT_TUPLE, _msgpack_packb(list(obj)))
+    raise TypeError(f"Cannot msgpack-encode {type(obj).__name__}")
+
+
+def _msgpack_ext_hook(code: int, data: bytes) -> Any:
+    if code == _EXT_TUPLE:
+        return tuple(_msgpack_unpackb(data))
+    return msgpack.ExtType(code, data)
+
+
+def _msgpack_packb(obj: Any) -> bytes:
+    # ``strict_types`` ensures tuples reach ``default`` instead of being
+    # auto-coerced to arrays; subclasses also route to the cloudpickle fallback.
+    return bytes(
+        msgpack.packb(obj, use_bin_type=True, strict_types=True, default=_msgpack_default)
+    )
+
+
+def _msgpack_unpackb(data: bytes) -> Any:
+    return msgpack.unpackb(data, raw=False, ext_hook=_msgpack_ext_hook)
 
 
 @runtime_checkable
@@ -47,22 +90,52 @@ def loads(self, data: bytes) -> Any:
 class MsgPackSerializer:
     """MsgPack-based serializer for compact, cross-language payloads.
 
-    Requires the ``msgpack`` extra::
-
-        pip install taskito[msgpack]
+    Only handles msgpack-native types. For arbitrary Python objects use
+    :class:`SmartSerializer`, which falls back to cloudpickle.
     """
 
     def dumps(self, obj: Any) -> bytes:
-        import msgpack
-
         return bytes(msgpack.packb(obj, use_bin_type=True))
 
     def loads(self, data: bytes) -> Any:
-        import msgpack
-
         return msgpack.unpackb(data, raw=False)
 
 
+class SmartSerializer:
+    """Default serializer: msgpack for plain payloads, cloudpickle fallback.
+
+    Plain data (the common case) serializes via msgpack — faster and more
+    compact than cloudpickle. Anything msgpack can't encode (lambdas, closures,
+    arbitrary class instances) transparently falls back to cloudpickle. A
+    one-byte tag records which codec produced each payload.
+
+    Tuples are preserved (via a msgpack ExtType), so payloads round-trip with
+    exact Python semantics.
+
+    Backward compatible: untagged payloads (written by older versions, raw
+    cloudpickle) are detected and loaded as cloudpickle.
+    """
+
+    def dumps(self, obj: Any) -> bytes:
+        try:
+            return _CODEC_MSGPACK + _msgpack_packb(obj)
+        except Exception:
+            # msgpack rejects non-native types (lambdas, custom classes, …);
+            # cloudpickle handles them.
+            return _CODEC_CLOUDPICKLE + bytes(cloudpickle.dumps(obj))
+
+    def loads(self, data: bytes) -> Any:
+        if not data:
+            raise ValueError("Cannot deserialize empty payload")
+        tag, body = data[:1], data[1:]
+        if tag == _CODEC_MSGPACK:
+            return _msgpack_unpackb(body)
+        if tag == _CODEC_CLOUDPICKLE:
+            return cloudpickle.loads(body)
+        # Untagged: a legacy cloudpickle payload from before the envelope existed.
+        return cloudpickle.loads(data)
+
+
 class EncryptedSerializer:
     """Wraps another serializer with AES-256-GCM encryption at rest.
 

pyproject.toml

@@ -24,6 +24,7 @@ classifiers = [
 ]
 dependencies = [
     "cloudpickle>=3.0",
+    "msgpack>=1.0",
     'tzdata; platform_system == "Windows"',
 ]
 [project.urls]

tests/core/test_serializers.py

@@ -5,7 +5,12 @@
 
 import pytest
 
-from taskito.serializers import CloudpickleSerializer, JsonSerializer, Serializer
+from taskito.serializers import (
+    CloudpickleSerializer,
+    JsonSerializer,
+    Serializer,
+    SmartSerializer,
+)
 
 
 class TestJsonSerializer:
@@ -148,3 +153,67 @@ def test_short_ciphertext_fails(self) -> None:
         s = EncryptedSerializer(JsonSerializer(), os.urandom(32))
         with pytest.raises(ValueError, match="too short"):
             s.loads(b"only-twelve-")
+
+
+class TestSmartSerializer:
+    def test_plain_data_uses_msgpack_tag(self) -> None:
+        s = SmartSerializer()
+        encoded = s.dumps({"x": 1, "y": ["a", "b"]})
+        assert encoded[:1] == b"\x01"
+        assert s.loads(encoded) == {"x": 1, "y": ["a", "b"]}
+
+    def test_namedtuple_preserved_via_cloudpickle(self) -> None:
+        """Tuple subclasses (namedtuples) must keep their type, not be
+        demoted to a plain tuple by the ExtType path."""
+        from collections import namedtuple
+
+        Point = namedtuple("Point", ["x", "y"])
+        s = SmartSerializer()
+        encoded = s.dumps(Point(1, 2))
+        assert encoded[:1] == b"\x00"  # cloudpickle fallback, not msgpack
+        restored = s.loads(encoded)
+        assert restored == Point(1, 2)
+        assert type(restored) is Point
+        assert restored.x == 1
+
+    def test_falls_back_to_cloudpickle_for_lambda(self) -> None:
+        s = SmartSerializer()
+        encoded = s.dumps(lambda x: x + 1)
+        assert encoded[:1] == b"\x00"
+        assert s.loads(encoded)(41) == 42
+
+    def test_falls_back_to_cloudpickle_for_custom_class(self) -> None:
+        s = SmartSerializer()
+
+        class Point:
+            def __init__(self, x: int) -> None:
+                self.x = x
+
+        encoded = s.dumps(Point(7))
+        assert encoded[:1] == b"\x00"
+        assert s.loads(encoded).x == 7
+
+    def test_loads_legacy_untagged_cloudpickle(self) -> None:
+        """Payloads written before the envelope existed are raw cloudpickle
+        and must still deserialize."""
+        legacy = CloudpickleSerializer().dumps({"legacy": True})
+        # A protocol-2+ pickle starts with 0x80, never a codec tag.
+        assert legacy[:1] not in (b"\x00", b"\x01")
+        assert SmartSerializer().loads(legacy) == {"legacy": True}
+
+    def test_preserves_tuples(self) -> None:
+        """Tuples round-trip as tuples (not lists) via the msgpack ExtType,
+        including nested ones — so task args/results keep exact semantics."""
+        s = SmartSerializer()
+        payload = ((1, "two", 3.0), {"k": ("nested", "tuple")})
+        restored = s.loads(s.dumps(payload))
+        assert restored == payload
+        assert isinstance(restored[0], tuple)
+        assert isinstance(restored[1]["k"], tuple)
+
+    def test_empty_payload_rejected(self) -> None:
+        with pytest.raises(ValueError, match="empty payload"):
+            SmartSerializer().loads(b"")
+
+    def test_satisfies_protocol(self) -> None:
+        assert isinstance(SmartSerializer(), Serializer)