Skip to content

Commit bfec2b1

Browse files
authored
fix(mcp): reset background capture after fork (#816)
1 parent 017349d commit bfec2b1

3 files changed

Lines changed: 104 additions & 0 deletions

File tree

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
pypi/posthog: patch
3+
---
4+
5+
Reset MCP background capture state after fork

posthog/mcp/_instrumentation.py

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111

1212
import asyncio
1313
import concurrent.futures
14+
import os
1415
import threading
1516
from datetime import datetime, timezone
1617
from typing import Any, Dict, List, Optional, Set
@@ -36,6 +37,23 @@
3637
_bg_loop_lock = threading.Lock()
3738

3839

40+
def _reinit_background_loop_after_fork() -> None:
41+
"""Drop background-loop state inherited by a forked child.
42+
43+
The loop's daemon thread does not survive ``fork()``, and its lock may have
44+
been held by a vanished thread. Replace the state without acquiring the old
45+
lock or trying to close the inherited loop, which can no longer be driven.
46+
"""
47+
global _BACKGROUND_TASKS, _bg_loop, _bg_loop_lock
48+
_BACKGROUND_TASKS = set()
49+
_bg_loop = None
50+
_bg_loop_lock = threading.Lock()
51+
52+
53+
if hasattr(os, "register_at_fork"):
54+
os.register_at_fork(after_in_child=_reinit_background_loop_after_fork)
55+
56+
3957
def _get_background_loop() -> asyncio.AbstractEventLoop:
4058
global _bg_loop
4159
if _bg_loop is None:
Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,81 @@
1+
import asyncio
2+
import os
3+
import signal
4+
import threading
5+
import warnings
6+
7+
import pytest
8+
9+
import posthog.mcp._instrumentation as instrumentation
10+
11+
12+
@pytest.mark.skipif(
13+
not hasattr(os, "fork") or not hasattr(os, "register_at_fork"),
14+
reason="requires os.fork and os.register_at_fork",
15+
)
16+
def test_sync_capture_completes_after_fork():
17+
parent_capture_started = threading.Event()
18+
finish_parent_capture = threading.Event()
19+
20+
async def pending_parent_capture():
21+
parent_capture_started.set()
22+
while not finish_parent_capture.is_set():
23+
await asyncio.sleep(0.01)
24+
25+
instrumentation.fire_and_forget(pending_parent_capture())
26+
assert parent_capture_started.wait(timeout=2)
27+
parent_loop = instrumentation._bg_loop
28+
assert parent_loop is not None
29+
assert instrumentation._BACKGROUND_TASKS
30+
31+
read_fd, write_fd = os.pipe()
32+
instrumentation._bg_loop_lock.acquire()
33+
try:
34+
with warnings.catch_warnings():
35+
warnings.simplefilter("ignore", DeprecationWarning)
36+
pid = os.fork()
37+
if pid == 0:
38+
os.close(read_fd)
39+
signal.alarm(5)
40+
try:
41+
inherited_work_cleared = not instrumentation._BACKGROUND_TASKS
42+
child_capture_completed = []
43+
44+
async def child_capture():
45+
child_capture_completed.append(True)
46+
47+
instrumentation.fire_and_forget(child_capture())
48+
instrumentation.drain_pending_sync(timeout=2)
49+
new_loop_created = instrumentation._bg_loop is not parent_loop
50+
51+
if (
52+
inherited_work_cleared
53+
and child_capture_completed == [True]
54+
and new_loop_created
55+
):
56+
result = "ok"
57+
else:
58+
result = (
59+
f"inherited_work_cleared={inherited_work_cleared}, "
60+
f"child_capture_completed={child_capture_completed}, "
61+
f"new_loop_created={new_loop_created}"
62+
)
63+
except BaseException as error:
64+
result = f"exception: {error!r}"
65+
finally:
66+
signal.alarm(0)
67+
os.write(write_fd, result.encode())
68+
os.close(write_fd)
69+
os._exit(0)
70+
71+
os.close(write_fd)
72+
result = os.read(read_fd, 4096).decode()
73+
os.close(read_fd)
74+
_, status = os.waitpid(pid, 0)
75+
finally:
76+
instrumentation._bg_loop_lock.release()
77+
finish_parent_capture.set()
78+
instrumentation.drain_pending_sync(timeout=2)
79+
80+
assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0, result
81+
assert result == "ok"

0 commit comments

Comments
 (0)