feat: each invoke is seeded differently

Author: brian14708

src/duron/_core/invoke.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import base64
+import binascii
 import contextlib
 import os
 import time
@@ -97,6 +98,7 @@ def get_init() -> InitParams:
                 "version": _CURRENT_VERSION,
                 "args": [codec.encode_json(arg) for arg in args],
                 "kwargs": {k: codec.encode_json(v) for k, v in kwargs.items()},
+                "seed": binascii.b2a_base64(os.urandom(12), newline=False).decode(),
             }
 
         codec = self._fn.codec
@@ -237,6 +239,7 @@ class InitParams(TypedDict):
     version: int
     args: list[JSONValue]
     kwargs: dict[str, JSONValue]
+    seed: str
 
 
 async def _invoke_prelude(
@@ -252,6 +255,7 @@ async def _invoke_prelude(
         if init_params["version"] != _CURRENT_VERSION:
             msg = "version mismatch"
             raise RuntimeError(msg)
+        loop.set_key(binascii.a2b_base64(init_params["seed"]))
         extra_kwargs: dict[str, object] = {}
         for name, type_, dtype in job_fn.inject:
             if type_ is Stream:
@@ -320,7 +324,7 @@ def __init__(
         ]
         | None = None,
     ) -> None:
-        self._loop = create_loop(asyncio.get_running_loop(), b"")
+        self._loop = create_loop(asyncio.get_running_loop())
         self._task = self._loop.create_task(task)
         self._log = log
         self._codec = codec

src/duron/_loop.py

@@ -92,24 +92,28 @@ class EventLoop(asyncio.AbstractEventLoop):
         "_timers",
     )
 
-    def __init__(self, host: asyncio.AbstractEventLoop, seed: bytes) -> None:
+    def __init__(self, host: asyncio.AbstractEventLoop) -> None:
         self._ready: deque[asyncio.Handle] = deque()
         self._debug: bool = False
         self._host: asyncio.AbstractEventLoop = host
         self._exc_handler: (
             Callable[[asyncio.AbstractEventLoop, dict[str, object]], object] | None
         ) = None
         self._ops: dict[bytes, OpFuture[object]] = {}
-        self._ctx: _TaskCtx = _TaskCtx(parent_id=seed)
+        self._ctx: _TaskCtx = _TaskCtx(parent_id=b"")
+        self._key: bytes = b""
         self._now_us: int = 0
         self._closed: bool = False
         self._event: asyncio.Event = asyncio.Event()  # loop = _host
         self._timers: list[asyncio.TimerHandle] = []
 
+    def set_key(self, key: bytes) -> None:
+        self._key = key
+
     def generate_op_id(self) -> bytes:
         ctx = _task_ctx.get(self._ctx)
         ctx.seq += 1
-        return _mix_id(ctx.parent_id, ctx.seq - 1)
+        return _mix_id(ctx.parent_id, self._key, ctx.seq - 1)
 
     def host_loop(self) -> asyncio.AbstractEventLoop:
         return self._host
@@ -284,7 +288,7 @@ def post_completion(
         if op := self._ops.pop(id_, None):
             if op.done():
                 return
-            tid = _mix_id(op.id, -1)
+            tid = _mix_id(op.id, self._key, -1)
             token = _task_ctx.set(_TaskCtx(parent_id=tid))
             if exception is None:
                 _ = self.call_soon(op.set_result, result)
@@ -352,17 +356,16 @@ def _timer_handle_cancelled(self, _th: asyncio.TimerHandle) -> None:
         pass
 
 
-def _mix_id(a: bytes, b: int) -> bytes:
+def _mix_id(a: bytes, key: bytes, b: int) -> bytes:
     if b == -1:
-        return blake2b(a, digest_size=12).digest()
-    return blake2b(b.to_bytes(4, "little") + a, digest_size=12).digest()
+        return blake2b(a, key=key, digest_size=12).digest()
+    return blake2b(b.to_bytes(4, "little") + a, key=key, digest_size=12).digest()
 
 
 def create_loop(
     parent_loop: asyncio.AbstractEventLoop,
-    seed: bytes,
 ) -> EventLoop:
-    return EventLoop(parent_loop, seed)  # type: ignore[abstract]
+    return EventLoop(parent_loop)  # type: ignore[abstract]
 
 
 def _copy_future_state(source: asyncio.Future[_T], dest: asyncio.Future[_T]) -> None:

tests/test_invoke.py

@@ -29,22 +29,10 @@ async def activity(ctx: Context, i: str) -> str:
         _ = await ctx.run(lambda: asyncio.sleep(0.1))
         return i + ":".join(x)
 
-    ids = {
-        "+qPYuDgKBdMkb8ME",
-        "20QZakraLA2aFVh0",
-        "9nLMU+itD7QHcCsf",
-        "BCLA1azFK4LrrEHg",
-        "D7qSBNZIThKa2P+H",
-        "d0eMKz6qFm4C+TYX",
-        "tlmy+UdKzdxIIykQ",
-        "vP2AH9GHFTbZRJQ7",
-    }
-
     log = MemoryLogStorage()
     async with activity.invoke(log) as t:
         await t.start("test")
         a = await t.wait()
-    assert {e["id"] for e in await log.entries()} == ids
 
     async with activity.invoke(log) as t:
         await t.start("test")
@@ -56,7 +44,6 @@ async def activity(ctx: Context, i: str) -> str:
         await t.start("test")
         c = await t.wait()
     assert a == c
-    assert {e["id"] for e in await log2.entries()} == ids
 
 
 @pytest.mark.asyncio
@@ -162,7 +149,6 @@ async def activity(ctx: Context) -> int:
         await t.start()
         a = await t.wait()
 
-    log = MemoryLogStorage()
     async with activity.invoke(log) as t:
         await t.start()
         b = await t.wait()
@@ -172,10 +158,12 @@ async def activity(ctx: Context) -> int:
 
 @pytest.mark.asyncio
 async def test_external_promise() -> None:
+    v: dict[str, str] = {}
+
     @fn
     async def activity(ctx: Context) -> int:
         a, b = await ctx.create_promise(int)
-        assert a == "9mcIBsvU2ej9uDsV"
+        v["data"] = a
         return await b
 
     log = MemoryLogStorage()
@@ -184,11 +172,11 @@ async def activity(ctx: Context) -> int:
 
         async def do() -> None:
             while True:
-                try:
-                    await t.complete_promise("9mcIBsvU2ej9uDsV", result=9)
-                    break
-                except ValueError:
-                    await asyncio.sleep(0.1)
+                if v.get("data") is None:
+                    await asyncio.sleep(0.01)
+                    continue
+                await t.complete_promise(v["data"], result=9)
+                break
 
         bg = asyncio.create_task(do())
         assert await t.wait() == 9

tests/test_loop.py

@@ -20,7 +20,7 @@ async def timer() -> int:
             )
         return 0
 
-    loop = create_loop(asyncio.get_event_loop(), b"")
+    loop = create_loop(asyncio.get_event_loop())
     loop.tick(time.time_ns())
     tsk = loop.create_task(timer())
     while (waitset := loop.poll_completion(tsk)) is not None:
@@ -45,7 +45,7 @@ async def op() -> None:
         _ = await loop.create_op(6)
 
     ids: set[bytes] = set()
-    loop = create_loop(asyncio.get_event_loop(), b"tsk")
+    loop = create_loop(asyncio.get_event_loop())
     loop.tick(time.time_ns())
     tsk = loop.create_task(op())
 
@@ -77,12 +77,12 @@ def tick(n: int, expect: set[int] | None) -> list[bytes] | None:
 @pytest.mark.asyncio
 async def test_op() -> None:  # noqa: RUF029
     baseline = {
-        b"\x89U\x82\xd9\xe9\xa1\x01\x0fb\xab}\xba",
-        b"\xa2VE\xcb\xf3\x81\x82\xe75@\xe9\xdf",
-        b"\xb2\x1d\xb3\xd5c\xe3J\no\xa6U\x18",
-        b"\xe7y\tDQ-\xe8\xfb\x9dBX\xde",
-        b"\t\xf9?\xf8kJ\xd4\ry8\xf2\xae",
-        b"\xc6\xbbEu\xdf\xd1uc\xf5M\x11'",
+        b'\x04"\xc0\xd5\xac\xc5+\x82\xeb\xacA\xe0',
+        b"\x07\x04%\x7f\xbf\xc3x*\x89}Jb",
+        b"\x8f\x1f\x1dMR\x19\xe7\xbf\xa2D\xbe\x7f",
+        b'\xb4"\xd9\x1a\x89\x896\xcb.\x9b#P',
+        b"\xf6g\x08\x06\xcb\xd4\xd9\xe8\xfd\xb8;\x15",
+        b"\xfa\xa3\xd8\xb88\n\x05\xd3$o\xc3\x04",
     }
     for _ in range(4):
         assert op_single() == baseline