From 55fcb7f977c2492fd53c2d02cb7f78865be77971 Mon Sep 17 00:00:00 2001 From: Enoch Tang Date: Wed, 28 Oct 2026 16:03:44 -0400 Subject: [PATCH] fix(client): Join split surrogate pairs when encoding task parameters A Python string can hold an emoji as two separate surrogate code points (for example after `ast.literal_eval` on escaped text). msgpack encodes strings as strict UTF-8, so `create_activation` raised `UnicodeEncodeError` and the caller's task failed. When msgpack raises `UnicodeEncodeError`, join the pairs back into one character and retry. Lone halves can't be rebuilt, so they become U+FFFD. The retry only runs after the error, so normal tasks are unchanged. Each retry increments `taskworker.producer.surrogates_repaired`. Refs STREAM-1999 Co-Authored-By: Claude --- clients/python/src/taskbroker_client/task.py | 37 +++++++++-- clients/python/tests/test_task.py | 65 ++++++++++++++++++++ 2 files changed, 96 insertions(+), 6 deletions(-) diff --git a/clients/python/src/taskbroker_client/task.py b/clients/python/src/taskbroker_client/task.py index 67d68d5f..c3429e54 100644 --- a/clients/python/src/taskbroker_client/task.py +++ b/clients/python/src/taskbroker_client/task.py @@ -40,6 +40,21 @@ R = TypeVar("R") +def _join_surrogates(value: Any) -> Any: + """ + Rebuild strings that hold a UTF-16 surrogate pair as two separate code points + (e.g. an emoji parsed with `ast.literal_eval`), which msgpack can't encode as + UTF-8. Pairs are joined into one character and lone halves become U+FFFD. + """ + if isinstance(value, str): + return value.encode("utf-16", "surrogatepass").decode("utf-16", "replace") + if isinstance(value, dict): + return {_join_surrogates(k): _join_surrogates(v) for k, v in value.items()} + if isinstance(value, (list, tuple)): + return [_join_surrogates(v) for v in value] + return value + + def assert_typed_kwarg( func: Callable[..., Any], param_name: str, @@ -290,6 +305,13 @@ def _signal_send(self, task: Task[Any, Any], args: Any, kwargs: Any) -> None: """ pass + def _producer_metric_tags(self) -> dict[str, str]: + return { + "namespace": self._namespace.name, + "taskname": self.name, + "topic": self._namespace.topic, + } + def create_activation( self, args: Collection[Any], @@ -342,7 +364,14 @@ def create_activation( ) data = {"args": args, "kwargs": kwargs} - msgpack_bytes = msgpack.packb(data, use_bin_type=True) + try: + msgpack_bytes = msgpack.packb(data, use_bin_type=True) + except UnicodeEncodeError: + msgpack_bytes = msgpack.packb(_join_surrogates(data), use_bin_type=True) + self.namespace.metrics.incr( + "taskworker.producer.surrogates_repaired", + tags=self._producer_metric_tags(), + ) should_compress = ( self.compression_type == CompressionType.ZSTD @@ -355,11 +384,7 @@ def create_activation( parameters_bytes_val = zstd.compress(msgpack_bytes) elapsed = time.perf_counter() - start_time - metric_tags = { - "namespace": self._namespace.name, - "taskname": self.name, - "topic": self._namespace.topic, - } + metric_tags = self._producer_metric_tags() self.namespace.metrics.distribution( "taskworker.producer.compressed_parameters_size", len(parameters_bytes_val), diff --git a/clients/python/tests/test_task.py b/clients/python/tests/test_task.py index b870f381..47110bcc 100644 --- a/clients/python/tests/test_task.py +++ b/clients/python/tests/test_task.py @@ -297,6 +297,71 @@ def with_parameters(one: str, two: int, org_id: int) -> None: assert activation.parameters == "" +EMOJI = chr(0x1F600) +# The same emoji as two separate surrogate code points, which isn't valid UTF-8. +SPLIT_EMOJI = chr(0xD83D) + chr(0xDE00) + + +def test_create_activation_joins_split_surrogate_pair(task_namespace: TaskNamespace) -> None: + @task_namespace.register(name="test.surrogates") + def with_message(message: str) -> None: + raise NotImplementedError + + with patch.object(task_namespace.metrics, "incr") as mock_incr: + activation = with_message.create_activation([], {"message": "hi " + SPLIT_EMOJI}) + + params = msgpack.unpackb(activation.parameters_bytes, raw=False) + assert params["kwargs"] == {"message": "hi " + EMOJI} + mock_incr.assert_called_once_with( + "taskworker.producer.surrogates_repaired", + tags={ + "namespace": task_namespace.name, + "taskname": "test.surrogates", + "topic": task_namespace.topic, + }, + ) + + +def test_create_activation_joins_nested_surrogate_pairs(task_namespace: TaskNamespace) -> None: + @task_namespace.register(name="test.surrogates_nested") + def with_nested(data: dict[str, Any], count: int) -> None: + raise NotImplementedError + + activation = with_nested.create_activation( + [{SPLIT_EMOJI: [SPLIT_EMOJI, b"raw", 1]}], + {"count": 2}, + ) + + params = msgpack.unpackb(activation.parameters_bytes, raw=False) + assert params["args"] == [{EMOJI: [EMOJI, b"raw", 1]}] + assert params["kwargs"] == {"count": 2} + + +def test_create_activation_replaces_lone_surrogate(task_namespace: TaskNamespace) -> None: + @task_namespace.register(name="test.surrogates_lone") + def with_message(message: str) -> None: + raise NotImplementedError + + activation = with_message.create_activation([], {"message": "hi " + chr(0xD83D)}) + + params = msgpack.unpackb(activation.parameters_bytes, raw=False) + assert params["kwargs"] == {"message": "hi " + chr(0xFFFD)} + + +def test_create_activation_valid_unicode_unchanged(task_namespace: TaskNamespace) -> None: + @task_namespace.register(name="test.unicode") + def with_message(message: str) -> None: + raise NotImplementedError + + kwargs = {"message": "hi " + EMOJI} + with patch.object(task_namespace.metrics, "incr") as mock_incr: + activation = with_message.create_activation([], kwargs) + + expected = msgpack.packb({"args": [], "kwargs": kwargs}, use_bin_type=True) + assert activation.parameters_bytes == expected + mock_incr.assert_not_called() + + def test_create_activation_tracing(task_namespace: TaskNamespace) -> None: @task_namespace.register(name="test.parameters") def with_parameters(one: str, two: int, org_id: int) -> None: