Ensure profiling messages can be serialized despite deep nesting

distributed/profile.py

@@ -155,7 +155,11 @@ def process(
     merge
     """
     if depth is None:
-        depth = sys.getrecursionlimit() - 50
+        # Cut off rather conservatively since the output of the profiling
+        # sometimes need to be recursed into as well, e.g. for serialization
+        # which can cause recursion errors later on since this can generate
+        # deeply nested dictionaries
+        depth = min(250, sys.getrecursionlimit() // 4)
     if depth <= 0:
         return None
     if any(frame.f_code.co_filename.endswith(o) for o in omit):

distributed/protocol/tests/test_protocol.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import sys
 from threading import Thread
 from time import sleep
 
@@ -393,3 +394,21 @@ def _(header, frames):
     header, frames = serialize(MyObj(), serializers=serializers)
     o = deserialize(header, frames)
     assert isinstance(o, MyObj)
+
+
+def test_deeply_nested_structures():
+    # These kind of deeply nested structures are generated in our profiling code
+    def gen_deeply_nested(depth):
+        msg = {}
+        d = msg
+        while depth:
+            depth -= 1
+            d["children"] = d = {}
+        return msg
+
+    msg = gen_deeply_nested(sys.getrecursionlimit() - 100)
+    with pytest.raises(TypeError, match="Could not serialize object"):
+        serialize(msg, on_error="raise")
+
+    msg = gen_deeply_nested(sys.getrecursionlimit() // 4)
+    assert isinstance(serialize(msg), tuple)

distributed/tests/test_profile.py

@@ -370,7 +370,7 @@ def f(i):
             else:
                 return f(i - 1)
 
-        f(sys.getrecursionlimit() - 40)
+        f(sys.getrecursionlimit() - 100)
         process(frame, None, state)
         merge(state, state, state)