fix eventlet issues with detached mode

[PythonFZ]

Summary by CodeRabbit

Release Notes

• Bug Fixes
  • Improved daemon process initialization to ensure stable operation.
  • Enhanced celery worker process handling for improved reliability.

_{✏️ Tip: You can customize this high-level summary in your review settings.}

[coderabbitai]

📝 Walkthrough

Walkthrough

Integrates eventlet's subprocess module to improve compatibility with daemonized processes. Adds eventlet hub reinitialization after forking in the daemon startup sequence and updates the Celery worker invocation to use eventlet-compatible subprocess handling instead of the standard library implementation.

Changes

Cohort / File(s)	Summary
Eventlet Hub Reinitialization src/zndraw/cli.py	Adds import of eventlet.hubs and calls eventlet.hubs.use_hub() after os.setsid() in daemonize function to ensure hub state is properly reinitialized post-fork and kqueue file descriptors survive the fork operation.
Eventlet-Compatible Subprocess src/zndraw/start_celery.py	Imports subprocess_orig from eventlet.green.subprocess and updates run_celery_worker() function to use subprocess_orig.Popen instead of standard subprocess.Popen, including return type annotation updates in signature and docstring.

Estimated code review effort

🎯 2 (Simple) | ⏱️ ~12 minutes

Poem

🐰 With eventlet's grace, the rabbit did sing,
Forks and hubs aligned in everything,
Green subprocess paths, so smooth and so right,
Daemonized dreams now work through the night! 🌙✨

🚥 Pre-merge checks | ✅ 3

✅ Passed checks (3 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title accurately describes the main changes, which address eventlet hub reinitialization issues when running in detached/daemonized mode.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing touches

• 📝 Generate docstrings

---

🧹 Recent nitpick comments

src/zndraw/start_celery.py (1)

13-28: Consider using subprocess.Popen for type annotations.

While subprocess_orig is the unpatched subprocess module, using it in type annotations may confuse type checkers since they expect subprocess.Popen. The actual returned object is a subprocess.Popen instance.

Suggested type annotation improvement

+import subprocess
 from eventlet.green.subprocess import subprocess_orig
 
 if t.TYPE_CHECKING:
     from zndraw.config import ZnDrawConfig
 
 
-def run_celery_worker(config: "ZnDrawConfig") -> subprocess_orig.Popen:
+def run_celery_worker(config: "ZnDrawConfig") -> subprocess.Popen[bytes]:
     """Run a celery worker with proper configuration.
     ...
     Returns
     -------
-    subprocess_orig.Popen
+    subprocess.Popen
         Running celery worker process.
     """

---

📜 Recent review details

Configuration used: defaults

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between a1b362b and ba8d8d8.

📒 Files selected for processing (2)

• src/zndraw/cli.py
• src/zndraw/start_celery.py

🧰 Additional context used

📓 Path-based instructions (1)

**/*.py

📄 CodeRabbit inference engine (AGENTS.md)

**/*.py: You MUST NEVER @pytest.mark.xfail or similar - all tests must pass!
All default values (e.g., camera, particles) must be defined exclusively within the Pydantic model. Do not scatter fallback logic throughout the codebase.
Do not perform null checks combined with hardcoded literals for default values. Rely entirely on the schema to populate default values during initialization.
You can not use LUA scripts with Redis!
If sensible, implement collections.abc interfaces for your classes, such as MutableMapping or MutableSequence.
Use numpy style docstrings in Python code.
Docstrings must be concise and to the point.
Use type hints wherever possible in Python. Use list[int|float] | None instead of t.Optional[t.List[int|float]]!
Imports should always be at the top of the file in Python.

Files:

• src/zndraw/cli.py
• src/zndraw/start_celery.py

🧬 Code graph analysis (1)

src/zndraw/start_celery.py (1)

src/zndraw/config.py (1)

• ZnDrawConfig (91-225)

⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)

• GitHub Check: pytest (3.13, ubuntu-latest)
• GitHub Check: pytest (3.12, ubuntu-latest)
• GitHub Check: pytest (3.11, ubuntu-latest)

🔇 Additional comments (3)

src/zndraw/start_celery.py (2)

83-95: LGTM! Correct use of unpatched subprocess for worker spawning.

Using subprocess_orig.Popen ensures the Celery worker is spawned via the original, unpatched subprocess module. This avoids potential issues with eventlet's cooperative threading interfering with the fork/exec of the worker process. The worker then initializes its own eventlet pool via --pool=eventlet.

---

7-7: Import subprocess_orig from eventlet is valid and correctly used.

The import of subprocess_orig from eventlet.green.subprocess is a documented pattern in eventlet that provides access to the original unpatched subprocess module. This is the correct approach to spawn the Celery worker subprocess without eventlet's monkey-patching interference, ensuring proper process creation. Type annotations and function usage are consistent and correct throughout the file.

src/zndraw/cli.py (1)

57-60: Eventlet hub reinitialization is necessary here, but note the guideline conflict.

The use_hub() call after os.setsid() is correct—it reinitializes the event loop in the child process to prevent sharing the parent's kqueue/epoll file descriptors. However, this import-inside-function pattern violates the coding guideline that states "Imports should always be at the top of the file in Python."

While post-fork hub reinitialization is a necessary and correct pattern (confirmed by eventlet documentation and usage in production projects like OpenStack), the guideline has no documented exception for this case. Either add an explicit exception to the import guideline for post-fork scenarios, or document this pattern as an accepted violation.

_{✏️ Tip: You can disable this entire section by setting review_details to false in your review settings.}

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[codecov-commenter]

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
1203	1	1202	1

View the top 2 failed test(s) by shortest run time

tests/test_lock_socket_events.py::test_idempotent_acquire_emits_refresh_event

Stack Traces | 2.53s run time

authenticated_session = ('http://127.0.0.1:49985', 'test-lock-socket', 'e09252d0-efbf-4f88-b118-304b08e5bb44', {'Authorization': 'Bearer eyJhb...X-Session-ID': 'e09252d0-efbf-4f88-b118-304b08e5bb44'}, <socketio.client.Client object at 0x7f883c5a3250>, 'test-user')

    def test_idempotent_acquire_emits_refresh_event(authenticated_session):
        """Test that re-acquiring a lock (idempotent) emits lock:update with action=refreshed."""
        server, room, session_id, auth_headers, sio, user_name = authenticated_session
    
        # Acquire lock first
        response = requests.post(
            f"{server}/api/rooms/{room}/locks/trajectory:meta/acquire",
            json={"msg": "Initial message"},
            headers={**auth_headers, "X-Session-ID": session_id},
        )
        assert response.status_code == 200
        lock_token = response.json()["lockToken"]
        assert response.json()["refreshed"] is False  # First acquire
    
        # Track received events (after initial acquire)
        received_events = []
    
        def on_lock_update(data):
            received_events.append(data)
    
        sio.on("lock:update", on_lock_update)
    
        # Re-acquire lock (idempotent) with new message
        response = requests.post(
            f"{server}/api/rooms/{room}/locks/trajectory:meta/acquire",
            json={"msg": "Updated message"},
            headers={**auth_headers, "X-Session-ID": session_id},
        )
        assert response.status_code == 200
        assert response.json()["success"] is True
        assert response.json()["refreshed"] is True  # Idempotent re-acquire
        assert response.json()["lockToken"] == lock_token  # Same token returned
    
        # Wait for socket event
        time.sleep(0.5)
    
        # Verify lock:update event was received with action=refreshed
>       assert len(received_events) == 1
E       AssertionError: assert 2 == 1
E        +  where 2 = len([{'action': 'acquired', 'holder': 'test-user', 'message': 'Initial message', 'roomId': 'test-lock-socket', ...}, {'action': 'refreshed', 'holder': 'test-user', 'message': 'Updated message', 'roomId': 'test-lock-socket', ...}])

tests/test_lock_socket_events.py:97: AssertionError

tests/test_lock_socket_events.py::test_lock_release_emits_lock_update

Stack Traces | 2.73s run time

authenticated_session = ('http://127.0.0.1:56875', 'test-lock-socket', '9d6a0fbe-46a9-49f2-b08e-23a26d1f86c4', {'Authorization': 'Bearer eyJhb...X-Session-ID': '9d6a0fbe-46a9-49f2-b08e-23a26d1f86c4'}, <socketio.client.Client object at 0x7f3a22293e50>, 'test-user')

    def test_lock_release_emits_lock_update(authenticated_session):
        """Test that releasing a lock emits lock:update event with action=released."""
        server, room, session_id, auth_headers, sio, user_name = authenticated_session
    
        # Acquire lock first
        response = requests.post(
            f"{server}/api/rooms/{room}/locks/trajectory:meta/acquire",
            json={"msg": "Test lock"},
            headers={**auth_headers, "X-Session-ID": session_id},
        )
        assert response.status_code == 200
        lock_token = response.json()["lockToken"]
    
        # Track received events (after initial acquire)
        received_events = []
    
        def on_lock_update(data):
            received_events.append(data)
    
        sio.on("lock:update", on_lock_update)
    
        # Release lock
        response = requests.post(
            f"{server}/api/rooms/{room}/locks/trajectory:meta/release",
            json={"lockToken": lock_token},
            headers={**auth_headers, "X-Session-ID": session_id},
        )
        assert response.status_code == 200
        assert response.json()["success"] is True
    
        # Wait for socket event
        time.sleep(0.5)
    
        # Verify lock:update event was received
>       assert len(received_events) == 1
E       AssertionError: assert 2 == 1
E        +  where 2 = len([{'action': 'acquired', 'holder': 'test-user', 'message': 'Test lock', 'roomId': 'test-lock-socket', ...}, {'action': 'released', 'holder': None, 'message': None, 'roomId': 'test-lock-socket', ...}])

tests/test_lock_socket_events.py:142: AssertionError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}