[FSSDK-11901] Fix concurrency bug in cmab service

.github/workflows/python.yml

@@ -65,10 +65,8 @@ jobs:
       fail-fast: false
       matrix:
         python-version:
-            - "pypy-3.8"
             - "pypy-3.9"
             - "pypy-3.10"
-            - "3.8"
             - "3.9"
             - "3.10"
             - "3.11"
@@ -93,10 +91,8 @@ jobs:
       fail-fast: false
       matrix:
         python-version:
-            - "pypy-3.8"
             - "pypy-3.9"
             - "pypy-3.10"
-            - "3.8"
             - "3.9"
             - "3.10"
             - "3.11"

optimizely/bucketer.py

@@ -22,7 +22,7 @@
 if version_info < (3, 8):
     from typing_extensions import Final
 else:
-    from typing import Final  # type: ignore
+    from typing import Final
 
 
 if TYPE_CHECKING:

optimizely/cmab/cmab_service.py

@@ -13,6 +13,7 @@
 import uuid
 import json
 import hashlib
+import threading
 
 from typing import Optional, List, TypedDict
 from optimizely.cmab.cmab_client import DefaultCmabClient
@@ -21,6 +22,8 @@
 from optimizely.project_config import ProjectConfig
 from optimizely.decision.optimizely_decide_option import OptimizelyDecideOption
 from optimizely import logger as _logging
+from optimizely.lib import pymmh3 as mmh3
+NUM_LOCK_STRIPES = 1000
 
 
 class CmabDecision(TypedDict):
@@ -52,10 +55,25 @@ def __init__(self, cmab_cache: LRUCache[str, CmabCacheValue],
         self.cmab_cache = cmab_cache
         self.cmab_client = cmab_client
         self.logger = logger
+        self.locks = [threading.Lock() for _ in range(NUM_LOCK_STRIPES)]
+
+    def _get_lock_index(self, user_id: str, rule_id: str) -> int:
+        """Calculate the lock index for a given user and rule combination."""
+        # Create a hash of user_id + rule_id for consistent lock selection
+        hash_input = f"{user_id}{rule_id}"
+        hash_value = mmh3.hash(hash_input, seed=0) & 0xFFFFFFFF  # Convert to unsigned
+        return hash_value % NUM_LOCK_STRIPES
 
     def get_decision(self, project_config: ProjectConfig, user_context: OptimizelyUserContext,
                      rule_id: str, options: List[str]) -> CmabDecision:
 
+        lock_index = self._get_lock_index(user_context.user_id, rule_id)
+        with self.locks[lock_index]:
+            return self._get_decision(project_config, user_context, rule_id, options)
+
+    def _get_decision(self, project_config: ProjectConfig, user_context: OptimizelyUserContext,
+                      rule_id: str, options: List[str]) -> CmabDecision:
+
         filtered_attributes = self._filter_attributes(project_config, user_context, rule_id)
 
         if OptimizelyDecideOption.IGNORE_CMAB_CACHE in options:

optimizely/decision/optimizely_decide_option.py

@@ -16,7 +16,7 @@
 if version_info < (3, 8):
     from typing_extensions import Final
 else:
-    from typing import Final  # type: ignore
+    from typing import Final
 
 
 class OptimizelyDecideOption:

optimizely/decision/optimizely_decision_message.py

@@ -16,7 +16,7 @@
 if version_info < (3, 8):
     from typing_extensions import Final
 else:
-    from typing import Final  # type: ignore
+    from typing import Final
 
 
 class OptimizelyDecisionMessage:

optimizely/entities.py

@@ -17,7 +17,7 @@
 if version_info < (3, 8):
     from typing_extensions import Final
 else:
-    from typing import Final  # type: ignore
+    from typing import Final
 
 
 if TYPE_CHECKING:

optimizely/event/event_factory.py

@@ -25,7 +25,7 @@
 if version_info < (3, 8):
     from typing_extensions import Final
 else:
-    from typing import Final  # type: ignore
+    from typing import Final
 
 if TYPE_CHECKING:
     # prevent circular dependenacy by skipping import at runtime

optimizely/event/event_processor.py

@@ -34,7 +34,7 @@
 if version_info < (3, 8):
     from typing_extensions import Final
 else:
-    from typing import Final  # type: ignore
+    from typing import Final
 
 
 class BaseEventProcessor(ABC):

optimizely/event/log_event.py

@@ -20,7 +20,7 @@
 if version_info < (3, 8):
     from typing_extensions import Literal
 else:
-    from typing import Literal  # type: ignore
+    from typing import Literal
 
 
 class LogEvent(event_builder.Event):

optimizely/event/user_event.py

@@ -22,7 +22,7 @@
 if version_info < (3, 8):
     from typing_extensions import Final
 else:
-    from typing import Final  # type: ignore
+    from typing import Final
 
 
 if TYPE_CHECKING:

optimizely/event_builder.py

@@ -25,7 +25,7 @@
 if version_info < (3, 8):
     from typing_extensions import Final, Literal
 else:
-    from typing import Final, Literal  # type: ignore
+    from typing import Final, Literal
 
 if TYPE_CHECKING:
     # prevent circular dependenacy by skipping import at runtime

optimizely/event_dispatcher.py

@@ -26,7 +26,7 @@
 if version_info < (3, 8):
     from typing_extensions import Protocol
 else:
-    from typing import Protocol  # type: ignore
+    from typing import Protocol
 
 
 class CustomEventDispatcher(Protocol):

optimizely/helpers/condition.py

@@ -32,7 +32,7 @@
 if version_info < (3, 8):
     from typing_extensions import Literal, Final
 else:
-    from typing import Literal, Final  # type: ignore
+    from typing import Literal, Final
 
 
 class ConditionOperatorTypes:

optimizely/helpers/enums.py

@@ -17,7 +17,7 @@
 if version_info < (3, 8):
     from typing_extensions import Final
 else:
-    from typing import Final  # type: ignore
+    from typing import Final
 
 
 class CommonAudienceEvaluationLogs:

optimizely/helpers/event_tag_utils.py

@@ -21,7 +21,7 @@
 if version_info < (3, 8):
     from typing_extensions import Final
 else:
-    from typing import Final  # type: ignore
+    from typing import Final
 
 
 if TYPE_CHECKING:

optimizely/logger.py

@@ -20,7 +20,7 @@
 if version_info < (3, 8):
     from typing_extensions import Final
 else:
-    from typing import Final  # type: ignore
+    from typing import Final
 
 
 _DEFAULT_LOG_FORMAT: Final = '%(levelname)-8s %(asctime)s %(filename)s:%(lineno)s:%(message)s'

optimizely/notification_center.py

@@ -20,7 +20,7 @@
 if version_info < (3, 8):
     from typing_extensions import Final
 else:
-    from typing import Final  # type: ignore
+    from typing import Final
 
 
 NOTIFICATION_TYPES: Final = tuple(

optimizely/odp/lru_cache.py

@@ -22,7 +22,7 @@
 if version_info < (3, 8):
     from typing_extensions import Protocol
 else:
-    from typing import Protocol  # type: ignore
+    from typing import Protocol
 
 # generic type definitions for LRUCache parameters
 K = TypeVar('K', bound=Hashable, contravariant=True)

optimizely/odp/optimizely_odp_option.py

@@ -16,7 +16,7 @@
 if version_info < (3, 8):
     from typing_extensions import Final
 else:
-    from typing import Final  # type: ignore
+    from typing import Final
 
 
 class OptimizelyOdpOption:

optimizely/project_config.py

@@ -24,7 +24,7 @@
 if version_info < (3, 8):
     from typing_extensions import Final
 else:
-    from typing import Final  # type: ignore
+    from typing import Final
 
 if TYPE_CHECKING:
     # prevent circular dependenacy by skipping import at runtime

optimizely/user_profile.py

@@ -19,7 +19,7 @@
 if version_info < (3, 8):
     from typing_extensions import Final
 else:
-    from typing import Final, TYPE_CHECKING  # type: ignore
+    from typing import Final, TYPE_CHECKING
 
     if TYPE_CHECKING:
         # prevent circular dependenacy by skipping import at runtime

tests/test_cmab_service.py

@@ -12,7 +12,7 @@
 # limitations under the License.
 import unittest
 from unittest.mock import MagicMock
-from optimizely.cmab.cmab_service import DefaultCmabService
+from optimizely.cmab.cmab_service import DefaultCmabService, NUM_LOCK_STRIPES
 from optimizely.optimizely_user_context import OptimizelyUserContext
 from optimizely.decision.optimizely_decide_option import OptimizelyDecideOption
 from optimizely.odp.lru_cache import LRUCache
@@ -185,3 +185,41 @@ def test_only_cmab_attributes_passed_to_client(self):
             {"age": 25, "location": "USA"},
             decision["cmab_uuid"]
         )
+
+    def test_same_user_rule_combination_uses_consistent_lock(self):
+        """Verifies that the same user/rule combination always uses the same lock index"""
+        user_id = "test_user"
+        rule_id = "test_rule"
+
+        # Get lock index multiple times
+        index1 = self.cmab_service._get_lock_index(user_id, rule_id)
+        index2 = self.cmab_service._get_lock_index(user_id, rule_id)
+        index3 = self.cmab_service._get_lock_index(user_id, rule_id)
+
+        # All should be the same
+        self.assertEqual(index1, index2, "Same user/rule should always use same lock")
+        self.assertEqual(index2, index3, "Same user/rule should always use same lock")
+
+    def test_lock_striping_distribution(self):
+        """Verifies that different user/rule combinations use different locks to allow for better concurrency"""
+        test_cases = [
+            ("user1", "rule1"),
+            ("user2", "rule1"),
+            ("user1", "rule2"),
+            ("user3", "rule3"),
+            ("user4", "rule4"),
+        ]
+
+        lock_indices = set()
+        for user_id, rule_id in test_cases:
+            index = self.cmab_service._get_lock_index(user_id, rule_id)
+
+            # Verify index is within expected range
+            self.assertGreaterEqual(index, 0, "Lock index should be non-negative")
+            self.assertLess(index, NUM_LOCK_STRIPES, "Lock index should be less than NUM_LOCK_STRIPES")
+
+            lock_indices.add(index)
+
+        # We should have multiple different lock indices (though not necessarily all unique due to hash collisions)
+        self.assertGreater(len(lock_indices), 1,
+                           "Different user/rule combinations should generally use different locks")