strict init timeout and some other fixes (#363)

- detailed reason return enum value instead of the enum
- remove uninitialized check that returns uninitialized eval details.
this early return omits possible info about local override/unsupported
config
- new option overall_init_timeout to time the entire init process
- moved anything that can throw during init to the class's initialize so
the class can still be constructed
- lock on some flakey unit tests
- add timeout to initialization details and error logs
- kong: statsig-io/kong#2786

Author: kat-statsig

statsig/evaluation_details.py

@@ -35,5 +35,5 @@ def __init__(self, config_sync_time: int, init_time: int,
 
     def detailed_reason(self):
         if self.reason == EvaluationReason.none:
-            return f"{self.source}"
-        return f"{self.source}:{self.reason}"
+            return f"{self.source.value}"
+        return f"{self.source.value}:{self.reason.value}"

statsig/evaluator.py

@@ -21,11 +21,14 @@ class _Evaluator:
     def __init__(self, spec_store: _SpecStore):
         self._spec_store = spec_store
 
-        self._country_lookup = CountryLookup()
+        self._country_lookup: Optional[CountryLookup] = None
         self._gate_overrides: Dict[str, dict] = {}
         self._config_overrides: Dict[str, dict] = {}
         self._layer_overrides: Dict[str, dict] = {}
 
+    def initialize(self):
+        self._country_lookup = CountryLookup()
+
     def override_gate(self, gate, value, user_id=None):
         gate_overrides = self._gate_overrides.get(gate)
         if gate_overrides is None:
@@ -195,9 +198,6 @@ def check_gate(self, user, gate, end_result=None, is_nested=False):
         if override is not None:
             return override
 
-        if self._spec_store.init_source == DataSource.UNINITIALIZED:
-            return _ConfigEvaluation(
-                evaluation_details=self._create_evaluation_details())
         eval_gate = self._spec_store.get_gate(gate)
         if eval_gate is None:
             globals.logger.debug(f"Gate {gate} not found in the store. Are you sure the gate name is correct?")
@@ -212,10 +212,6 @@ def get_config(self, user, config_name, is_exp=False):
         if override is not None:
             return override
 
-        if self._spec_store.init_source == DataSource.UNINITIALIZED:
-            return _ConfigEvaluation(
-                evaluation_details=self._create_evaluation_details())
-
         eval_config = self._spec_store.get_config(config_name)
         if eval_config is None:
             globals.logger.debug(
@@ -231,10 +227,6 @@ def get_layer(self, user, layer_name):
         if override is not None:
             return override
 
-        if self._spec_store.init_source == DataSource.UNINITIALIZED:
-            return _ConfigEvaluation(evaluation_details=
-                                     self._create_evaluation_details())
-
         eval_layer = self._spec_store.get_layer(layer_name)
         if eval_layer is None:
             globals.logger.debug(f"Layer {layer_name} not found in the store. Are you sure the layer name is correct?")
@@ -376,6 +368,8 @@ def __evaluate_condition(self, user, condition, end_result):
             if value is None:
                 ip = self.__get_from_user(user, "ip")
                 if ip is not None and field == "country":
+                    if not self._country_lookup:
+                        self._country_lookup = CountryLookup()
                     value = self._country_lookup.lookupStr(ip)
             if value is None:
                 return False

statsig/spec_store.py

@@ -6,7 +6,7 @@
 
 from . import globals
 from .constants import Const
-from .diagnostics import Context, Diagnostics, Marker
+from .diagnostics import Diagnostics, Marker
 from .evaluation_details import EvaluationReason, DataSource
 from .sdk_configs import _SDK_Configs
 from .spec_updater import SpecUpdater
@@ -83,7 +83,6 @@ def initialize(self):
         )
 
         self.spec_updater.download_id_lists(for_initialize=True)
-
         self.spec_updater.start_background_threads()
         self.spec_updater.initialized = True
 
@@ -343,13 +342,6 @@ def _log_process(self, msg, process=None):
             process = "Initialize" if not self.spec_updater.initialized else "Config Sync"
         globals.logger.log_process(process, msg)
 
-    def _get_current_context(self):
-        return (
-            Context.INITIALIZE
-            if not self.spec_updater.initialized
-            else Context.CONFIG_SYNC
-        )
-
     def _get_initialize_strategy(self) -> List[DataSource]:
         try:
             if self._options.initialize_sources is not None:

statsig/spec_updater.py

@@ -43,7 +43,7 @@ def __init__(
         self._statsig_metadata = statsig_metadata
         self._background_download_configs = None
         self._background_download_id_lists = None
-        self._config_sync_strategies = self._get_sync_dcs_strategies()
+        self._config_sync_strategies = [DataSource.NETWORK]
         self._dcs_process_lock = threading.Lock()
         if options.out_of_sync_threshold_in_s is not None:
             self._enforce_sync_fallback_threshold_in_ms: Optional[float] = options.out_of_sync_threshold_in_s * 1000
@@ -56,6 +56,7 @@ def __init__(
         self.dcs_listener: Optional[Callable] = None
         self.id_lists_listener: Optional[Callable] = None
         self.data_adapter = data_adapter
+        self._config_sync_strategies = self._get_sync_dcs_strategies()
 
     def get_config_spec(self, source: DataSource, for_initialize=False):
         try:

statsig/statsig_context.py

@@ -10,10 +10,12 @@ class InitContext:
     error: Optional[Exception]
     source: DataSource
     store_populated: bool
+    timed_out: bool
 
     def __init__(self):
         self.start_time = int(time.time() * 1000)
         self.success = False
         self.error = None
         self.source = DataSource.UNINITIALIZED
         self.store_populated = False
+        self.timed_out = False

statsig/statsig_errors.py

@@ -8,3 +8,7 @@ class StatsigRuntimeError(RuntimeError):
 
 class StatsigNameError(NameError):
     pass
+
+
+class StatsigTimeoutError(TimeoutError):
+    pass

statsig/statsig_network.py

@@ -80,7 +80,7 @@ def __init__(
     ):
         self.sdk_key = sdk_key
         self.error_boundary = error_boundary
-        self.statsig_options = options
+        self.options = options
         self.diagnostics = diagnostics
         self.shutdown_event = shutdown_event
         self.statsig_metadata = statsig_metadata
@@ -91,7 +91,7 @@ def __init__(
         self.id_list_worker: IStatsigNetworkWorker = defaultHttpWorker
         self.log_event_worker: IStatsigNetworkWorker = defaultHttpWorker
         self.http_worker: IStatsigNetworkWorker = defaultHttpWorker
-        for endpoint, config in options.proxy_configs.items():
+        for endpoint, config in self.options.proxy_configs.items():
             protocol = config.protocol
             if protocol == NetworkProtocol.GRPC:
                 self.load_grpc_worker(endpoint, config)
@@ -107,7 +107,7 @@ def load_grpc_websocket_worker(self, endpoint: NetworkEndpoint, config: ProxyCon
         grpc_webhook_worker = grpc_webhook_worker_class(
             self.sdk_key,
             config,
-            self.statsig_options,
+            self.options,
             self.error_boundary,
             self.diagnostics,
             self.shutdown_event,
@@ -156,7 +156,7 @@ def get_dcs(
             log_on_exception: Optional[bool] = False,
             init_timeout: Optional[int] = None,
     ):
-        if self.statsig_options.local_mode:
+        if self.options.local_mode:
             globals.logger.warning("Local mode is enabled. Not fetching DCS.")
             return
         self.dcs_worker.get_dcs(on_complete, since_time, log_on_exception, init_timeout)
@@ -168,14 +168,14 @@ def get_dcs_fallback(
             log_on_exception: Optional[bool] = False,
             init_timeout: Optional[int] = None,
     ):
-        if self.statsig_options.local_mode:
+        if self.options.local_mode:
             globals.logger.warning("Local mode is enabled. Not fetching DCS with fallback.")
             return
-        dcs_proxy = self.statsig_options.proxy_configs.get(NetworkEndpoint.DOWNLOAD_CONFIG_SPECS)
+        dcs_proxy = self.options.proxy_configs.get(NetworkEndpoint.DOWNLOAD_CONFIG_SPECS)
         is_proxy_dcs = (
                 dcs_proxy
                 and dcs_proxy.proxy_address != STATSIG_CDN
-                or self.statsig_options.api_for_download_config_specs != STATSIG_CDN
+                or self.options.api_for_download_config_specs != STATSIG_CDN
         )
         if is_proxy_dcs:
             self.http_worker.get_dcs_fallback(on_complete, since_time, log_on_exception, init_timeout)
@@ -186,7 +186,7 @@ def get_id_lists(
             log_on_exception: Optional[bool] = False,
             init_timeout: Optional[int] = None,
     ):
-        if self.statsig_options.local_mode:
+        if self.options.local_mode:
             globals.logger.warning("Local mode is enabled. Not fetching ID Lists.")
             return
         self.id_list_worker.get_id_lists(on_complete, log_on_exception, init_timeout)
@@ -197,42 +197,42 @@ def get_id_lists_fallback(
             log_on_exception: Optional[bool] = False,
             init_timeout: Optional[int] = None,
     ):
-        if self.statsig_options.local_mode:
+        if self.options.local_mode:
             globals.logger.warning("Local mode is enabled. Not fetching ID Lists with fallback.")
             return
-        if not self.statsig_options.fallback_to_statsig_api:
+        if not self.options.fallback_to_statsig_api:
             return
-        id_list_proxy = self.statsig_options.proxy_configs.get(
+        id_list_proxy = self.options.proxy_configs.get(
             NetworkEndpoint.GET_ID_LISTS
         )
-        id_list_api_override = self.statsig_options.api_for_get_id_lists
+        id_list_api_override = self.options.api_for_get_id_lists
         is_id_lists_proxy = id_list_api_override != STATSIG_API or (
                 id_list_proxy and id_list_proxy.proxy_address != STATSIG_API)
         if is_id_lists_proxy:
             self.http_worker.get_id_lists_fallback(on_complete, log_on_exception, init_timeout)
 
     def get_id_list(self, on_complete: Any, url, headers, log_on_exception=False):
-        if self.statsig_options.local_mode:
+        if self.options.local_mode:
             globals.logger.warning("Local mode is enabled. Not fetching ID List.")
             return
         self.http_worker.get_id_list(on_complete, url, headers, log_on_exception)
 
     def log_events(self, payload, headers=None, log_on_exception=False, retry=0):
-        if self.statsig_options.local_mode:
+        if self.options.local_mode:
             globals.logger.warning("Local mode is enabled. Not logging events.")
             return None
         return self.log_event_worker.log_events(
             payload, headers=headers, log_on_exception=log_on_exception, retry=retry
         )
 
     def listen_for_dcs(self, listeners: IStreamingListeners, fallback: Callable):
-        if self.statsig_options.local_mode:
+        if self.options.local_mode:
             globals.logger.warning("Local mode is enabled. Not listening for DCS.")
             return
         if isinstance(self.dcs_worker, IStatsigWebhookWorker):
             self.dcs_worker.start_listen_for_config_spec(listeners)
             interval = (
-                    self.statsig_options.rulesets_sync_interval
+                    self.options.rulesets_sync_interval
                     or DEFAULT_RULESET_SYNC_INTERVAL
             )
             callbacks = StreamingFallback(
@@ -244,7 +244,7 @@ def listen_for_dcs(self, listeners: IStreamingListeners, fallback: Callable):
             self.dcs_worker.register_fallback_cb(callbacks)
 
     def listen_for_id_lists(self, listeners: IStreamingListeners):
-        if self.statsig_options.local_mode:
+        if self.options.local_mode:
             globals.logger.warning("Local mode is enabled. Not listening for ID Lists.")
             return
         if isinstance(self.id_list_worker, IStatsigWebhookWorker):

statsig/statsig_options.py

@@ -107,6 +107,7 @@ def __init__(
             initialize_sources: Optional[List[DataSource]] = None,
             config_sync_sources: Optional[List[DataSource]] = None,
             output_logger_level: Optional[LogLevel] = LogLevel.WARNING,
+            overall_init_timeout: Optional[float] = None,
     ):
         self.data_store = data_store
         self._environment: Union[None, dict] = None
@@ -152,6 +153,7 @@ def __init__(
         self.initialize_sources = initialize_sources
         self.config_sync_sources = config_sync_sources
         self.output_logger_level = output_logger_level
+        self.overall_init_timeout = overall_init_timeout
         self._logging_copy: Dict[str, Any] = {}
         self._set_logging_copy()
         self._attributes_changed = False
@@ -218,5 +220,7 @@ def _set_logging_copy(self):
             logging_copy["event_queue_size"] = self.event_queue_size
         if self.retry_queue_size != DEFAULT_RETRY_QUEUE_SIZE:
             logging_copy["retry_queue_size"] = self.retry_queue_size
+        if self.overall_init_timeout is not None:
+            logging_copy["overall_init_timeout"] = self.overall_init_timeout
         self._logging_copy = logging_copy
         self._attributes_changed = False

statsig/statsig_server.py

@@ -1,6 +1,7 @@
 import dataclasses
 import threading
 import time
+from concurrent.futures import ThreadPoolExecutor
 from typing import Optional, Union, Tuple
 
 from . import globals
@@ -15,7 +16,7 @@
 from .spec_store import _SpecStore, EntityType
 from .statsig_context import InitContext
 from .statsig_error_boundary import _StatsigErrorBoundary
-from .statsig_errors import StatsigNameError, StatsigRuntimeError, StatsigValueError
+from .statsig_errors import StatsigNameError, StatsigRuntimeError, StatsigValueError, StatsigTimeoutError
 from .statsig_event import StatsigEvent
 from .statsig_logger import _StatsigLogger
 from .statsig_metadata import _StatsigMetadata
@@ -37,14 +38,16 @@ class InitializeDetails:
     init_success: bool
     store_populated: bool
     error: Optional[Exception]
+    timed_out: bool
 
     def __init__(self, duration: int, source: str, init_success: bool, store_populated: bool,
-                 error: Optional[Exception]):
+                 error: Optional[Exception], timed_out: bool = False):
         self.duration = duration
         self.source = source
         self.init_success = init_success
         self.error = error
         self.store_populated = store_populated
+        self.timed_out = timed_out
 
 
 class StatsigServer:
@@ -61,6 +64,7 @@ class StatsigServer:
     _evaluator: _Evaluator
 
     def __init__(self) -> None:
+        self._executor = ThreadPoolExecutor(max_workers=1)
         self._initialized = False
 
         self._errorBoundary = _StatsigErrorBoundary()
@@ -109,12 +113,15 @@ def _post_init_logging(self, options: StatsigOptions, init_details: InitializeDe
                 globals.logger.error(
                     "Statsig SDK instance initialized, but config store is not populated. The SDK is using default values for evaluation.")
         else:
-            globals.logger.error("Statsig SDK instance Initialized failed!")
+            if init_details.timed_out:
+                globals.logger.error("Statsig SDK instance initialization timed out.")
+            else:
+                globals.logger.error("Statsig SDK instance Initialized failed!")
 
     def _initialize_impl(self, sdk_key: str, options: Optional[StatsigOptions]):
         threw_error = False
+        init_context = InitContext()
         try:
-            init_context = InitContext()
             diagnostics = Diagnostics()
             diagnostics.add_marker(Marker().overall().start())
 
@@ -157,13 +164,28 @@ def _initialize_impl(self, sdk_key: str, options: Optional[StatsigOptions]):
             self._evaluator = _Evaluator(self._spec_store)
             self._sampling_key_set = TTLSet(self.__shutdown_event)
 
-            self._spec_store.initialize()
-            self._initialized = True
+            init_timeout = options.overall_init_timeout
+            if init_timeout is not None:
+                future = self._executor.submit(self._initialize_instance)
+                try:
+                    future.result(timeout=init_timeout)
+                except Exception as e:
+                    raise StatsigTimeoutError() from e
+            else:
+                self._initialize_instance()
 
         except (StatsigValueError, StatsigNameError, StatsigRuntimeError) as e:
             threw_error = True
             raise e
 
+        except StatsigTimeoutError:
+            globals.logger.warning(
+                "Statsig SDK instance initialization timed out. Initializing store in the background.")
+            self._executor.submit(self._initialize_instance)
+            threw_error = True
+            init_context.timed_out = True
+            self._initialized = True
+
         except Exception as e:
             threw_error = True
             self._errorBoundary.log_exception("initialize", e)
@@ -176,7 +198,13 @@ def _initialize_impl(self, sdk_key: str, options: Optional[StatsigOptions]):
             init_context.success = not threw_error
 
         return InitializeDetails(int(time.time() * 1000) - init_context.start_time, init_context.source,
-                                 init_context.success, init_context.store_populated, init_context.error)
+                                 init_context.success, init_context.store_populated, init_context.error,
+                                 init_context.timed_out)
+
+    def _initialize_instance(self):
+        self._evaluator.initialize()
+        self._spec_store.initialize()
+        self._initialized = True
 
     def is_store_populated(self):
         try:
@@ -423,6 +451,7 @@ def flush(self):
 
     def shutdown(self):
         def task():
+            globals.logger.info("Shutting down Statsig SDK instance.")
             self.__shutdown_event.set()
             self._logger.shutdown()
             self._spec_store.shutdown()