fix: Remove get_cached() and use get() everywhere for cross-worker visibility

src/ogx/core/routers/vector_io.py

@@ -87,18 +87,15 @@ async def shutdown(self) -> None:
         logger.debug("VectorIORouter.shutdown")
         pass
 
-    def _get_provider_id(self, vector_store_id: str) -> str:
+    async def _get_provider_id(self, vector_store_id: str) -> str:
         """Get the provider ID for a vector store for metrics labeling (best-effort).
 
-        Uses the same in-memory cache (get_cached) that the routing table's
-        get_provider_impl uses when dispatching operations, so this does NOT
-        cause an extra DB/async lookup on the hot path.
-
-        Returns "unknown" only as a fallback so that a metrics-label lookup
-        failure never blocks the actual operation.
-        """
+        Uses dist_registry.get() (cache-then-DB) so that multi-worker deployments
+        can resolve provider IDs for vector stores created by other workers. Falls
+        back to "unknown" on failure so a metrics-label lookup never blocks the
+        actual operation."""
         try:
-            obj = self.routing_table.dist_registry.get_cached("vector_store", vector_store_id)
+            obj = await self.routing_table.dist_registry.get("vector_store", vector_store_id)
             if obj is None:
                 logger.warning("Vector store not found in registry cache", vector_store_id=vector_store_id)
                 return "unknown"
@@ -174,7 +171,7 @@ async def insert_chunks(
         )
         start_time = time.perf_counter()
         num_chunks = len(request.chunks)
-        provider_id = self._get_provider_id(request.vector_store_id)
+        provider_id = await self._get_provider_id(request.vector_store_id)
         metric_attrs = create_vector_metric_attributes(
             vector_db=request.vector_store_id,
             operation="chunks",
@@ -208,7 +205,7 @@ async def query_chunks(
     ) -> QueryChunksResponse:
         logger.debug("VectorIORouter.query_chunks", vector_store_id=request.vector_store_id)
         start_time = time.perf_counter()
-        provider_id = self._get_provider_id(request.vector_store_id)
+        provider_id = await self._get_provider_id(request.vector_store_id)
         metric_attrs = create_vector_metric_attributes(
             vector_db=request.vector_store_id,
             operation="query",
@@ -450,7 +447,7 @@ async def openai_delete_vector_store(
         vector_store_id: str,
     ) -> VectorStoreDeleteResponse:
         logger.debug("VectorIORouter.openai_delete_vector_store", vector_store_id=vector_store_id)
-        provider_id = self._get_provider_id(vector_store_id)
+        provider_id = await self._get_provider_id(vector_store_id)
         metric_attrs = create_vector_metric_attributes(
             vector_db=vector_store_id,
             operation="store",
@@ -474,7 +471,7 @@ async def openai_search_vector_store(
     ) -> VectorStoreSearchResponsePage:
         logger.debug("VectorIORouter.openai_search_vector_store", vector_store_id=vector_store_id)
         start_time = time.perf_counter()
-        provider_id = self._get_provider_id(vector_store_id)
+        provider_id = await self._get_provider_id(vector_store_id)
         search_mode = getattr(request, "search_mode", "vector")
         metric_attrs = create_vector_metric_attributes(
             vector_db=vector_store_id,
@@ -532,7 +529,7 @@ async def openai_attach_file_to_vector_store(
             file_id=request.file_id,
         )
         start_time = time.perf_counter()
-        provider_id = self._get_provider_id(vector_store_id)
+        provider_id = await self._get_provider_id(vector_store_id)
         metric_attrs = create_vector_metric_attributes(
             vector_db=vector_store_id,
             operation="attach",
@@ -649,7 +646,7 @@ async def openai_delete_vector_store_file(
         file_id: str,
     ) -> VectorStoreFileDeleteResponse:
         logger.debug("VectorIORouter.openai_delete_vector_store_file", vector_store_id=vector_store_id, file_id=file_id)
-        provider_id = self._get_provider_id(vector_store_id)
+        provider_id = await self._get_provider_id(vector_store_id)
         metric_attrs = create_vector_metric_attributes(
             vector_db=vector_store_id,
             operation="file",

src/ogx/core/routing_tables/common.py

@@ -144,8 +144,11 @@ def apiname_object():
 
         apiname, objtype = apiname_object()
 
-        # Get objects from disk registry
-        obj = self.dist_registry.get_cached(objtype, routing_key)
+        # Get objects from disk registry (via cache with DB fallback for
+        # multi-worker scenarios). Using get() rather than a cache-only
+        # accessor ensures we see objects created by other workers even
+        # before the TTL refresh fires.
+        obj = await self.dist_registry.get(objtype, routing_key)
         if not obj:
             provider_ids = list(self.impls_by_provider_id.keys())
             if len(provider_ids) > 1:

src/ogx/core/store/registry.py

@@ -29,8 +29,6 @@ async def initialize(self) -> None: ...
 
     async def get(self, type: str, identifier: str) -> RoutableObjectWithProvider | None: ...
 
-    def get_cached(self, type: str, identifier: str) -> RoutableObjectWithProvider | None: ...
-
     async def update(self, obj: RoutableObjectWithProvider) -> RoutableObjectWithProvider: ...
 
     async def register(self, obj: RoutableObjectWithProvider) -> bool: ...
@@ -72,10 +70,6 @@ def __init__(self, kvstore: KVStore):
     async def initialize(self) -> None:
         pass
 
-    def get_cached(self, type: str, identifier: str) -> RoutableObjectWithProvider | None:
-        # Disk registry does not have a cache
-        raise NotImplementedError("Disk registry does not have a cache")
-
     async def get_all(self) -> list[RoutableObjectWithProvider]:
         start_key, end_key = _get_registry_key_range()
         values = await self.kvstore.values_in_range(start_key, end_key)
@@ -182,9 +176,6 @@ async def _ensure_initialized(self):
     async def initialize(self) -> None:
         await self._ensure_initialized()
 
-    def get_cached(self, type: str, identifier: str) -> RoutableObjectWithProvider | None:
-        return self.cache.get((type, identifier), None)
-
     def _should_refresh_cache(self) -> bool:
         """Check if cache should be refreshed based on TTL."""
         current_time = time.time()
@@ -238,10 +229,10 @@ async def get(self, type: str, identifier: str) -> RoutableObjectWithProvider |
 
     async def register(self, obj: RoutableObjectWithProvider) -> bool:
         await self._ensure_initialized()
-        # Use super().get() (DB read) rather than self.get_cached() so that in
-        # multi-worker deployments, where each process has its own in-memory
-        # cache, we always read the authoritative stored object regardless of
-        # whether this worker's cache was warmed by _ensure_initialized().
+        # Read from disk (not cache) to handle multi-worker scenarios where
+        # each process has its own in-memory cache. This ensures we always
+        # get the authoritative stored object regardless of whether this
+        # worker's cache was warmed by _ensure_initialized().
         existing_obj = await super().get(obj.type, obj.identifier)
         success = await super().register(obj)
 

tests/unit/core/routers/test_vector_stores_abac.py

@@ -50,7 +50,7 @@ class MockDistRegistry:
     def __init__(self):
         self.dist = None
 
-    def get_cached(self, type_name: str, identifier: str):
+    async def get(self, type_name: str, identifier: str):
         return None
 
     async def register(self, obj):

tests/unit/registry/test_registry.py

@@ -5,9 +5,12 @@
 # the root directory of this source tree.
 
 
+from unittest.mock import Mock
+
 import pytest
 
 from ogx.core.datatypes import User, VectorStoreWithOwner
+from ogx.core.routers.vector_io import VectorIORouter
 from ogx.core.storage.datatypes import SqliteKVStoreConfig
 from ogx.core.storage.kvstore.sqlite.sqlite import SqliteKVStoreImpl
 from ogx.core.store.registry import (
@@ -96,9 +99,8 @@ async def test_cached_registry_updates(cached_disk_dist_registry):
     )
     await cached_disk_dist_registry.register(new_vector_store)
 
-    # Verify in cache — covers the else-obj branch: on first registration
-    # (no existing DB object) the incoming obj itself must be cached.
-    cached_vector_store = cached_disk_dist_registry.get_cached("vector_store", "test_vector_store_2")
+    # Verify in cache
+    cached_vector_store = cached_disk_dist_registry.cache.get(("vector_store", "test_vector_store_2"))
     assert cached_vector_store is not None
     assert cached_vector_store.identifier == new_vector_store.identifier
     assert cached_vector_store.provider_id == new_vector_store.provider_id
@@ -368,7 +370,7 @@ async def test_double_registration_with_cache_conflict(cached_disk_dist_registry
     assert result1 is True
 
     # Verify in cache
-    cached_model = cached_disk_dist_registry.get_cached("model", "test_model")
+    cached_model = cached_disk_dist_registry.cache.get(("model", "test_model"))
     assert cached_model is not None
     assert cached_model.model_type == ModelType.llm
 
@@ -377,7 +379,7 @@ async def test_double_registration_with_cache_conflict(cached_disk_dist_registry
         await cached_disk_dist_registry.register(model2)
 
     # Cache should still contain original model
-    cached_model_after = cached_disk_dist_registry.get_cached("model", "test_model")
+    cached_model_after = cached_disk_dist_registry.cache.get(("model", "test_model"))
     assert cached_model_after is not None
     assert cached_model_after.model_type == ModelType.llm
 
@@ -416,7 +418,7 @@ async def test_multi_worker_cache_synchronization(sqlite_kvstore, sample_vector_
     assert result_b.embedding_model == sample_vector_store.embedding_model
 
     # After the first get, Worker B should have it in cache
-    cached_b = worker_b_registry.get_cached("vector_store", "test_vector_store")
+    cached_b = worker_b_registry.cache.get(("vector_store", "test_vector_store"))
     assert cached_b is not None
     assert cached_b.identifier == sample_vector_store.identifier
 
@@ -457,7 +459,7 @@ async def test_cached_registry_preserves_owner_on_subset_reregistration(cached_d
     assert await cached_disk_dist_registry.register(subset_vs)
 
     # Cache must hold the full DB object — owner must not be dropped
-    cached = cached_disk_dist_registry.get_cached("vector_store", "owned_vs")
+    cached = cached_disk_dist_registry.cache.get(("vector_store", "owned_vs"))
     assert cached is not None
     assert cached.owner is not None
     assert cached.owner.principal == "admin"
@@ -495,3 +497,44 @@ async def test_multi_worker_get_all_synchronization(sqlite_kvstore, sample_vecto
     identifiers_b = {obj.identifier for obj in all_b}
     assert "test_vector_store" in identifiers_b
     assert "test_model" in identifiers_b
+
+
+async def test_get_provider_id_cross_worker_visibility(sqlite_kvstore, sample_vector_store):
+    """Test that two routers sharing a database can resolve provider IDs across workers.
+
+    This exercises the cross-worker path through VectorIORouter._get_provider_id():
+    1. Worker A registers a vector store
+    2. Worker B's router resolves the same vector store's provider ID
+       via dist_registry.get() cache-then-DB fallback
+
+    Before the fix (when callers used get_cached()), Worker B would return
+    None or "unknown" because its in-memory cache had no record. After the
+    fix, get() falls back to the shared database.
+    """
+    # Two separate registries simulating two uvicorn workers with separate
+    # in-memory caches but sharing the same SQLite backend
+    worker_a_registry = CachedDiskDistributionRegistry(sqlite_kvstore, cache_ttl_seconds=0)
+    await worker_a_registry.initialize()
+
+    worker_b_registry = CachedDiskDistributionRegistry(sqlite_kvstore, cache_ttl_seconds=0)
+    await worker_b_registry.initialize()
+
+    # Workers get their own routing tables, each pointing to its own registry
+    a_rt = Mock(dist_registry=worker_a_registry)
+    b_rt = Mock(dist_registry=worker_b_registry)
+    a_router = VectorIORouter(a_rt)
+    b_router = VectorIORouter(b_rt)
+
+    # Worker A registers the vector store
+    await worker_a_registry.register(sample_vector_store)
+
+    # Worker A can see its provider immediately
+    provider_a = await a_router._get_provider_id("test_vector_store")
+    assert provider_a == sample_vector_store.provider_id
+
+    # Worker B's cache is empty but can still resolve via DB fallback
+    provider_b = await b_router._get_provider_id("test_vector_store")
+    assert provider_b == sample_vector_store.provider_id
+
+    # Both must agree on the provider
+    assert provider_a == provider_b

tests/unit/registry/test_registry_acl.py

@@ -22,7 +22,7 @@ async def test_registry_cache_with_acl(cached_disk_dist_registry):
     success = await cached_disk_dist_registry.register(model)
     assert success
 
-    cached_model = cached_disk_dist_registry.get_cached("model", "model-acl")
+    cached_model = cached_disk_dist_registry.cache.get(("model", "model-acl"))
     assert cached_model is not None
     assert cached_model.identifier == "model-acl"
     assert cached_model.owner.principal == "testuser"
@@ -56,7 +56,7 @@ async def test_registry_empty_acl(cached_disk_dist_registry):
 
     await cached_disk_dist_registry.register(model)
 
-    cached_model = cached_disk_dist_registry.get_cached("model", "model-empty-acl")
+    cached_model = cached_disk_dist_registry.cache.get(("model", "model-empty-acl"))
     assert cached_model is not None
     assert cached_model.owner is not None
     assert cached_model.owner.attributes is None
@@ -73,7 +73,7 @@ async def test_registry_empty_acl(cached_disk_dist_registry):
 
     await cached_disk_dist_registry.register(model)
 
-    cached_model = cached_disk_dist_registry.get_cached("model", "model-no-acl")
+    cached_model = cached_disk_dist_registry.cache.get(("model", "model-no-acl"))
     assert cached_model is not None
     assert cached_model.owner is None