Fix concurrent access bug

Author: cmanallen

sentry_sdk/feature_flags.py

@@ -1,5 +1,7 @@
+import copy
 import sentry_sdk
 from sentry_sdk._lru_cache import LRUCache
+from threading import Lock
 
 from typing import TYPE_CHECKING
 
@@ -16,20 +18,43 @@ class FlagBuffer:
 
     def __init__(self, capacity):
         # type: (int) -> None
-        self.buffer = LRUCache(capacity)
         self.capacity = capacity
+        self.lock = Lock()
+
+        # Buffer is private. The name is mangled to discourage use. If you use this attribute
+        # directly you're on your own!
+        self.__buffer = LRUCache(capacity)
 
     def clear(self):
         # type: () -> None
-        self.buffer = LRUCache(self.capacity)
+        self.__buffer = LRUCache(self.capacity)
+
+    def __deepcopy__(self, memo):
+        with self.lock:
+            buffer = FlagBuffer(self.capacity)
+            buffer.__buffer = copy.deepcopy(self.__buffer, memo)
+            return buffer
 
     def get(self):
         # type: () -> list[FlagData]
-        return [{"flag": key, "result": value} for key, value in self.buffer.get_all()]
+        with self.lock:
+            return [
+                {"flag": key, "result": value} for key, value in self.__buffer.get_all()
+            ]
 
     def set(self, flag, result):
         # type: (str, bool) -> None
-        self.buffer.set(flag, result)
+        if isinstance(result, FlagBuffer):
+            # If someone were to `self` in `self` this would create a circular dependency on
+            # the lock. This is of course a deadlock. However, this is far outside the expected
+            # usage of this class. We guard against it here for completeness and to document this
+            # expected failure mode.
+            raise ValueError(
+                "FlagBuffer instances can not be inserted into the dictionary."
+            )
+
+        with self.lock:
+            self.__buffer.set(flag, result)
 
 
 def add_feature_flag(flag, result):

tests/test_feature_flags.py

@@ -1,5 +1,7 @@
 import concurrent.futures as cf
 import sys
+import copy
+import threading
 
 import pytest
 
@@ -167,3 +169,35 @@ def test_flag_tracking():
         {"flag": "e", "result": False},
         {"flag": "f", "result": False},
     ]
+
+
+def test_flag_buffer_concurrent_access():
+    buffer = FlagBuffer(capacity=100)
+    error_occurred = False
+
+    def writer():
+        for i in range(1_000_000):
+            buffer.set(f"key_{i}", True)
+
+    def reader():
+        nonlocal error_occurred
+
+        try:
+            for _ in range(1000):
+                copy.deepcopy(buffer)
+        except RuntimeError:
+            error_occurred = True
+
+    writer_thread = threading.Thread(target=writer)
+    reader_thread = threading.Thread(target=reader)
+
+    writer_thread.start()
+    reader_thread.start()
+
+    writer_thread.join(timeout=5)
+    reader_thread.join(timeout=5)
+
+    # This should always be false. If this ever fails we know we have concurrent access to a
+    # shared resource. When deepcopying we should have exclusive access to the underlying
+    # memory.
+    assert error_occurred is False