fix: render a task's code the same way in every process

python-client/wmill/tests/test_workflow.py

@@ -1888,3 +1888,43 @@ class TestTaskFingerprint:
 
         result = _run_workflow(wf, {}, {})
         assert result["steps"][0]["fn_id"] != result["steps"][1]["fn_id"]
+
+    def test_two_lambdas_differing_inside_a_genexp_are_told_apart(self):
+        """The difference lives in a nested code object, and they share a source
+        line, so the shape has to be read recursively."""
+        first, second = task(lambda xs: sum(x + 1 for x in xs)), task(lambda xs: sum(x + 2 for x in xs))
+
+        @workflow
+        async def wf():
+            return await asyncio.gather(first(xs=[1]), second(xs=[1]))
+
+        result = _run_workflow(wf, {}, {})
+        assert result["steps"][0]["fn_id"] != result["steps"][1]["fn_id"]
+
+    def test_fingerprint_does_not_move_with_the_interpreter_hash_seed(self):
+        """A set constant renders in hash order, which the interpreter randomizes
+        per process: a fingerprint that moved with it would never hit its cache."""
+        import os
+        import pathlib
+        import subprocess
+        import sys
+
+        script = (
+            "from wmill.client import _fn_fingerprint\n"
+            "def t(x):\n"
+            "    return x in frozenset({'a', 'b', 'c', 'd', 'e'})\n"
+            "print(_fn_fingerprint(t))\n"
+        )
+        root = str(pathlib.Path(__file__).resolve().parents[1])
+        seen = set()
+        for seed in ("1", "2"):
+            env = {**os.environ, "PYTHONHASHSEED": seed, "PYTHONPATH": root}
+            out = subprocess.run(
+                [sys.executable, "-c", script],
+                capture_output=True,
+                text=True,
+                env=env,
+                check=True,
+            )
+            seen.add(out.stdout.strip())
+        assert len(seen) == 1

python-client/wmill/wmill/client.py

@@ -3285,6 +3285,36 @@ class WorkflowCtx:
         })
 
 
+def _code_shape(code) -> bytes:
+    """What a code object does, rendered the same way in every process.
+
+    A set renders in hash order, which the interpreter randomizes per process, so
+    a fingerprint built on it would change between jobs and never find its cached
+    result; sorting the members fixes that. A nested code object is where a
+    generator expression's body lives, and two of those can be the only
+    difference between two functions, so it is rendered rather than skipped.
+    """
+    import types
+
+    def const(c) -> bytes:
+        if isinstance(c, types.CodeType):
+            return b"code:" + _code_shape(c)
+        if isinstance(c, (frozenset, set)):
+            return b"set:" + b",".join(sorted(const(x) for x in c))
+        if isinstance(c, tuple):
+            return b"tuple:" + b",".join(const(x) for x in c)
+        return f"{type(c).__name__}:{c!r}".encode()
+
+    return b"|".join(
+        [
+            code.co_code,
+            repr(code.co_names).encode(),
+            repr(code.co_varnames).encode(),
+            b",".join(const(c) for c in code.co_consts),
+        ]
+    )
+
+
 def _fn_fingerprint(func) -> str:
     """A stable identity for a task's code, what its cached result is keyed on: a
     name is shared by any two tasks called the same, and a step key by any two
@@ -3303,8 +3333,7 @@ def _fn_fingerprint(func) -> str:
         pass
     code = getattr(func, "__code__", None)
     if code is not None:
-        consts = tuple(c for c in code.co_consts if not isinstance(c, type(code)))
-        parts.append(repr((code.co_code, code.co_names, code.co_varnames, consts)).encode())
+        parts.append(_code_shape(code))
     if not parts:
         parts.append(repr(func).encode())
     return hashlib.sha256(b"\x1f".join(parts)).hexdigest()