BUG: recover_sub_pool no longer raises on concurrent create_actor

python/xoscar/backends/indigen/pool.py

@@ -501,7 +501,24 @@ async def recover_sub_pool(self, address: str):
 
         if self._auto_recover == "actor":
             # need to recover all created actors
-            for _, message in self._allocated_actors[address].values():
+            # copy to a list first: create_actor() mutates this same dict
+            # concurrently, and iterating it directly across the `await` below
+            # can raise "dictionary changed size during iteration"
+            #
+            # skip in-flight placeholder entries (keyed by None):
+            # MainActorPoolBase.create_actor() inserts one before awaiting
+            # the sub pool create and only pops it once that call returns,
+            # so a placeholder can still be sitting in the dict here if the
+            # sub pool died mid create. Replaying it would resend the
+            # caller's original, unfinished CreateActorMessage
+            # (from_main=False) straight to the sub pool, which duplicates
+            # the still-pending create and can raise ActorAlreadyExist or
+            # register the actor a second time via notify_main_pool_to_create.
+            for actor_ref, (_, message) in list(
+                self._allocated_actors[address].items()
+            ):
+                if actor_ref is None:
+                    continue
                 create_actor_message: CreateActorMessage = message  # type: ignore
                 await self.call(address, create_actor_message)
 

python/xoscar/backends/indigen/tests/test_pool.py

@@ -16,11 +16,13 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import logging
 import multiprocessing
 import os
 import re
 import sys
+import threading
 import time
 from unittest import mock
 
@@ -59,6 +61,7 @@
     ErrorMessage,
     HasActorMessage,
     MessageType,
+    ResultMessage,
     SendMessage,
     TellMessage,
     new_message_id,
@@ -930,6 +933,138 @@ def on_process_recover(*_):
                 await ctx.has_actor(actor_ref)
 
 
+@pytest.mark.asyncio
+async def test_recover_sub_pool_concurrent_create_actor():
+    # GH-48: recover_sub_pool() iterated self._allocated_actors[address]
+    # directly across an await; a concurrent create_actor() mutating that
+    # same dict raised "dictionary changed size during iteration".
+    pool = object.__new__(MainActorPool)
+    pool.external_address = "dummy://main"
+    pool._config = mock.Mock(get_process_index=mock.Mock(return_value=0))
+    pool.sub_processes = {}
+    pool._auto_recover = "actor"
+    pool._allocation_lock = threading.Lock()
+    pool.start_sub_pool = mock.AsyncMock(return_value=None)
+    pool.wait_sub_pools_ready = mock.AsyncMock(return_value=(["proc"], ["addr"]))
+
+    sub_address = "dummy://sub"
+    existing_message = CreateActorMessage(
+        new_message_id(),
+        TestActor,
+        b"existing",
+        (),
+        {},
+        allocate_strategy=AddressSpecified(sub_address),
+    )
+    pool._allocated_actors = {
+        sub_address: {b"existing": (AddressSpecified(sub_address), existing_message)}
+    }
+
+    reached_recover_call = asyncio.Event()
+    reached_create_call = asyncio.Event()
+    release_recover_call = asyncio.Event()
+    call_count = 0
+
+    async def fake_call(address, message):
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            # recover_sub_pool()'s call for the pre-existing actor
+            reached_recover_call.set()
+            await release_recover_call.wait()
+        else:
+            # the racing create_actor() call; hang so its placeholder
+            # entry stays in the dict while recover_sub_pool resumes
+            reached_create_call.set()
+            await asyncio.Event().wait()
+        return ResultMessage(new_message_id(), b"actor_ref")
+
+    pool.call = fake_call
+
+    recover_task = asyncio.create_task(pool.recover_sub_pool(sub_address))
+    await asyncio.wait_for(reached_recover_call.wait(), timeout=5)
+
+    new_message = CreateActorMessage(
+        new_message_id(),
+        TestActor,
+        b"new",
+        (),
+        {},
+        allocate_strategy=AddressSpecified(sub_address),
+    )
+    create_task_ = asyncio.create_task(pool.create_actor(new_message))
+    await asyncio.wait_for(reached_create_call.wait(), timeout=5)
+    # create_actor() has already inserted its placeholder into the same
+    # dict recover_sub_pool() is mid-iteration over
+    assert len(pool._allocated_actors[sub_address]) == 2
+
+    release_recover_call.set()
+    await asyncio.wait_for(recover_task, timeout=5)
+
+    create_task_.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await create_task_
+
+
+@pytest.mark.asyncio
+async def test_recover_sub_pool_skips_in_flight_placeholder():
+    # GH-48 review follow-up: create_actor() inserts a placeholder under
+    # key None before awaiting the sub pool create and only removes it
+    # once that call returns; if the sub pool dies mid create, the
+    # placeholder is left behind holding the caller's original,
+    # unfinished CreateActorMessage. recover_sub_pool() must skip that
+    # placeholder rather than replay it, since replaying resends an
+    # unfinished create straight to the sub pool.
+    pool = object.__new__(MainActorPool)
+    pool.external_address = "dummy://main"
+    pool._config = mock.Mock(get_process_index=mock.Mock(return_value=0))
+    pool.sub_processes = {}
+    pool._auto_recover = "actor"
+    pool._allocation_lock = threading.Lock()
+    pool.start_sub_pool = mock.AsyncMock(return_value=None)
+    pool.wait_sub_pools_ready = mock.AsyncMock(return_value=(["proc"], ["addr"]))
+
+    sub_address = "dummy://sub"
+    existing_message = CreateActorMessage(
+        new_message_id(),
+        TestActor,
+        b"existing",
+        (),
+        {},
+        allocate_strategy=AddressSpecified(sub_address),
+    )
+    # a still in-flight create_actor() call left its placeholder (key
+    # None) behind because the sub pool died before it could be popped
+    placeholder_message = CreateActorMessage(
+        new_message_id(),
+        TestActor,
+        b"in-flight",
+        (),
+        {},
+        allocate_strategy=AddressSpecified(sub_address),
+    )
+    pool._allocated_actors = {
+        sub_address: {
+            b"existing": (AddressSpecified(sub_address), existing_message),
+            None: (AddressSpecified(sub_address), placeholder_message),
+        }
+    }
+
+    replayed_actor_ids = []
+
+    async def fake_call(address, message):
+        replayed_actor_ids.append(message.actor_id)
+        return ResultMessage(new_message_id(), b"actor_ref")
+
+    pool.call = fake_call
+
+    await pool.recover_sub_pool(sub_address)
+
+    # only the real, completed entry is replayed; the placeholder is
+    # skipped, never resent to the sub pool
+    assert replayed_actor_ids == [b"existing"]
+
+
 @pytest.mark.parametrize(
     "exception_config",
     [