fix(sdk-python): fetch prompts again in a forked process

docs/sdk-conformance.md

@@ -142,6 +142,11 @@ Concurrent calls share one fetch.
   and not to the full response. Only an address lookup can continue after the call stops waiting,
   until the system resolver gives up. Both SDKs measure the TTL and Retry-After in elapsed time, so
   a change of the system clock does not move them.
+- Fork (Python): a process that `os.fork()` makes gets the copies of the parent, but not the fetch
+  threads of the parent. A call that starts in the child does not wait for a fetch of the parent.
+  If the call must fetch, it starts a new fetch. If a signal handler forks while a call waits for a
+  fetch, that call continues in the child. The call gets `timeout` 10 seconds after the start of
+  the fetch, because no thread of the child completes the fetch.
 - Failure: when a fetch fails, the SDKs keep the last copy for another TTL and send the error to
   `onError` / `on_error`. With no copy, the call gets the fallback. With no fallback, the call gets
   a `PromptError`. With no copy, the calls of the next 10 seconds (or the TTL, if it is shorter)

sdks/python/README.md

@@ -219,6 +219,13 @@ pinned = telemetry_dev.get_prompt("support/reply", version=13)
   call returns it at once and fetches a new copy in a background thread. Only a call with no copy
   waits for a fetch, for 10 seconds at most, also when the network stops. Calls for the same
   prompt share one fetch. `get_prompt_async` runs the same function in a worker thread.
+- A process that `os.fork()` makes keeps the copies of its parent. This includes a process of
+  `multiprocessing` with the `fork` start method. A call that starts in the child does not wait
+  for a fetch of the parent. If the call must fetch, it starts a new fetch.
+- If a signal handler forks while a call waits for a fetch, that call continues in the child. No
+  thread of the child completes that fetch. Thus the call waits until 10 seconds after the start
+  of the fetch. Then the call returns the `fallback`, or raises `PromptError` with the code
+  `timeout`.
 - If a fetch fails, `get_prompt` returns the last copy that it fetched and reports the error
   through `on_error`. With no copy, it returns the `fallback` (`version` is `None` and
   `is_fallback` is `True`). With no fallback, it raises `PromptError` whose `code` tells why, for

sdks/python/src/telemetry_dev/_prompts.py

@@ -5,6 +5,7 @@
 import copy
 import hashlib
 import json
+import os
 import random
 import re
 import socket
@@ -455,12 +456,57 @@ class _Entry:
     retry_at: float = 0.0
     # Set when the running fetch ends.
     pending: threading.Event | None = None
+    # The process whose thread runs the fetch. A forked process gets the entry, but not the thread.
+    pending_pid: int = 0
     # When calls stop waiting for the running fetch.
     deadline: float = 0.0
 
 
 _cache: OrderedDict[tuple[str, str, str, str], _Entry] = OrderedDict()
-_cache_lock = threading.Lock()
+# Reentrant, so that a fork in a thread that holds the lock, for example in a signal handler, does
+# not wait for itself in _lock_for_fork. _reset_in_child can replace it, so read it from the module
+# each time.
+_cache_lock = threading.RLock()
+# Whether _lock_for_fork got the lock in this thread. os.fork() runs the after hooks also when a
+# before hook raises, for example in a signal handler.
+_forking = threading.local()
+
+
+def _lock_for_fork() -> None:
+    _cache_lock.acquire()
+    _forking.locked = True
+
+
+def _unlock_after_fork() -> None:
+    if getattr(_forking, "locked", False):
+        _forking.locked = False
+        _cache_lock.release()
+
+
+def _reset_in_child() -> None:
+    """Unlock the cache, and drop the fetches of the parent: the child has none of its threads."""
+    global _cache_lock
+    if getattr(_forking, "locked", False):
+        _forking.locked = False
+        _cache_lock.release()
+    else:
+        # The before hook did not get the lock, because a signal stopped its wait, or did not run,
+        # as in uWSGI with py-call-osafterfork. A thread of the parent can hold the lock, and only
+        # this thread exists here.
+        _cache_lock = threading.RLock()
+    for entry in _cache.values():
+        entry.pending = None
+
+
+# A fork that runs these hooks waits for a thread that holds the lock, so that the child gets no
+# entry that a thread changed only in part. os.register_at_fork cannot remove a hook, and
+# importlib.reload runs this module again with the same globals. Thus the hooks of the first run
+# stay, and they find the lock and the cache when they run. Windows has no fork.
+if hasattr(os, "register_at_fork") and "_fork_hooks" not in globals():
+    _fork_hooks = True
+    os.register_at_fork(
+        before=_lock_for_fork, after_in_parent=_unlock_after_fork, after_in_child=_reset_in_child
+    )
 
 
 def _load(
@@ -481,15 +527,23 @@ def _load(
             entry = _cache[key] = _Entry()
         now = monotonic()
         ttl = ttl_seconds if entry.response is not None else min(ttl_seconds, _ERROR_TTL_S)
-        if now >= max(entry.checked_at + ttl, entry.retry_at) and entry.pending is None:
-            entry.pending = threading.Event()
+        if entry.pending is not None and entry.pending_pid != os.getpid():
+            # A fork in C code that runs no hooks copied the fetch but not its thread. This check
+            # does not replace the lock. If a thread of the parent held the lock at that fork, a
+            # call in the child waits for the lock with no limit.
+            entry.pending = None
+        # A local, because a fork in this thread, as in a signal handler, can clear entry.pending.
+        pending = entry.pending
+        if now >= max(entry.checked_at + ttl, entry.retry_at) and pending is None:
+            pending = entry.pending = threading.Event()
+            entry.pending_pid = os.getpid()
             entry.deadline = now + _FETCH_TIMEOUT_S
             threading.Thread(
-                target=_refresh, args=(entry, entry.pending, fetch, on_error), daemon=True
+                target=_refresh, args=(entry, pending, fetch, on_error), daemon=True
             ).start()
         if entry.response is not None:
             return entry.response
-        pending, deadline = entry.pending, entry.deadline
+        deadline = entry.deadline
     if pending is not None and not pending.wait(deadline - monotonic()):
         raise _timeout()
     with _cache_lock:

sdks/python/tests/otlp_capture.py

@@ -33,7 +33,8 @@ class CaptureServer:
     status: int = 200
     # The (status, JSON body) replies to GET requests, in order; the last one repeats.
     replies: list[tuple[int, object]] = field(default_factory=lambda: [])
-    # Seconds that the server waits before it answers a GET request.
+    # Seconds that the server waits before it answers a GET request. The server reads it before the
+    # request shows in requests, so that a test that sees the request can change it for the next.
     delay: float = 0
     # The location header of GET replies, for a redirect status.
     location: str | None = None
@@ -66,14 +67,15 @@ def do_POST(self) -> None:
             self.end_headers()
 
         def do_GET(self) -> None:
+            delay = capture.delay
             captured.append(
                 CapturedRequest(
                     path=self.requestline.split(" ")[1],
                     headers={key.lower(): value for key, value in self.headers.items()},
                     body=b"",
                 )
             )
-            time.sleep(capture.delay)
+            time.sleep(delay)
             status, reply = (
                 capture.replies[0] if len(capture.replies) == 1 else capture.replies.pop(0)
             )