[FSSDK-11901] Fix concurrency bug in cmab service (#462)

* Update: Implement locking mechanism for CMAB service to enhance concurrency

* Update: Refactor get_decision method

* update: type checking fix

* Update: Remove Python 3.8 from testing matrix in CI workflow

Author: FarhanAnjum-opti

.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: