RDBC-1035 RDBC-1036 Add aggressive cache context managers and store-level events

redknightlois · redknightlois · commit fc03aba00b30 · 2026-04-10T10:49:14.000-03:00
- Add add/remove_before_store, add/remove_after_save_changes,
  add/remove_before_delete, add/remove_before_query to DocumentStoreBase;
  register_events_for_session wires them into every new session
- Add aggressively_cache_for(duration, mode) and disable_aggressive_caching()
  context managers on DocumentStore; AggressiveCacheOptions is thread-local
  via threading.local(), mirroring C# AsyncLocal&lt;T&gt;
- Add _AggressiveCacheEviction inner class: subscribes to Changes() feed and
  increments HttpCache.generation on DocumentChange / IndexChange events
- Fix _listen_to_changes_and_update_cache: construct _AggressiveCacheEviction
  outside __add_change_lock to prevent deadlock when store.changes() (which
  acquires the same lock) is called during construction
- Remove dead HttpCache.ReleaseCacheItem inner class; fix _get_from_cache
  to import and return the module-level ReleaseCacheItem for non-cacheable
  command paths (was AttributeError at runtime)
- Fix before_delete event timing for key-based session.delete(): move
  BeforeDeleteEventArgs dispatch from delete() into prepare_for_save_changes()
  via __prepare_for_key_deletes, matching C# PrepareForEntitiesDeletion timing
  and making both delete paths (key vs entity) consistent
- Fix session-level remove_before_query: was appending instead of removing,
  making it impossible to unregister a query event handler on an open session
- Rename session-level remove_before_delete_entity to remove_before_delete,
  matching the naming convention of all other remove methods and the store-level API
- Add no_caching guard to _check_aggressive_cache_guard test helper so it
  exactly mirrors the condition in execute(); add test for no_caching bypass
- Add integration tests verifying event args contents: args.entity identity
  in OnBeforeStore and args.key in OnBeforeDelete (per C# Events.cs spec)
- Remove test for unreachable RAW-command cache-hit path: _get_from_cache
  never returns a populated item for non-OBJECT commands, so the scenario
  the test described could not occur in production
diff --git a/ravendb/documents/session/document_session_operations/in_memory_document_session_operations.py b/ravendb/documents/session/document_session_operations/in_memory_document_session_operations.py
@@ -542,14 +542,14 @@ def remove_after_save_changes(self, event: Callable[[AfterSaveChangesEventArgs],
     def add_before_delete(self, event: Callable[[BeforeDeleteEventArgs], None]):
         self._before_delete.append(event)
 
-    def remove_before_delete_entity(self, event: Callable[[BeforeDeleteEventArgs], None]):
+    def remove_before_delete(self, event: Callable[[BeforeDeleteEventArgs], None]):
         self._before_delete.remove(event)
 
     def add_before_query(self, event: Callable[[BeforeQueryEventArgs], None]):
         self._before_query.append(event)
 
     def remove_before_query(self, event: Callable[[BeforeQueryEventArgs], None]):
-        self._before_query.append(event)
+        self._before_query.remove(event)
 
     def before_store_invoke(self, before_store_event_args: BeforeStoreEventArgs):
         for event in self._before_store:
diff --git a/ravendb/documents/store/definition.py b/ravendb/documents/store/definition.py
@@ -175,6 +175,30 @@ def operations(self) -> OperationExecutor:
     def open_session(self, database: Optional[str] = None, session_options: Optional = None):
         pass
 
+    def add_before_store(self, event: Callable[[BeforeStoreEventArgs], None]):
+        self.__before_store.append(event)
+
+    def remove_before_store(self, event: Callable[[BeforeStoreEventArgs], None]):
+        self.__before_store.remove(event)
+
+    def add_after_save_changes(self, event: Callable[[AfterSaveChangesEventArgs], None]):
+        self.__after_save_changes.append(event)
+
+    def remove_after_save_changes(self, event: Callable[[AfterSaveChangesEventArgs], None]):
+        self.__after_save_changes.remove(event)
+
+    def add_before_delete(self, event: Callable[[BeforeDeleteEventArgs], None]):
+        self.__before_delete.append(event)
+
+    def remove_before_delete(self, event: Callable[[BeforeDeleteEventArgs], None]):
+        self.__before_delete.remove(event)
+
+    def add_before_query(self, event: Callable[[BeforeQueryEventArgs], None]):
+        self.__before_query.append(event)
+
+    def remove_before_query(self, event: Callable[[BeforeQueryEventArgs], None]):
+        self.__before_query.remove(event)
+
     def add_on_session_creation(self, event: Callable[[SessionCreatedEventArgs], None]):
         self.__on_session_creation.append(event)
 
@@ -313,7 +337,7 @@ def __init__(self, urls: Union[str, List[str]] = None, database: Optional[str] =
         self.urls = [urls] if isinstance(urls, str) else urls
         self.database = database
         self.__request_executors: Dict[str, Lazy[RequestExecutor]] = CaseInsensitiveDict()
-        # todo: aggressive cache
+        self.__aggressive_cache_changes: Dict[str, "DocumentStore._AggressiveCacheInvalidator"] = {}
         self.__maintenance_operation_executor: Optional[MaintenanceOperationExecutor] = None
         self.__operation_executor: Optional[OperationExecutor] = None
         # todo: database smuggler
@@ -379,7 +403,9 @@ def close(self):
         for event in self.__before_close:
             event()
 
-        # todo: evict items from cache based on changes
+        for eviction in list(self.__aggressive_cache_changes.values()):
+            eviction.close()
+        self.__aggressive_cache_changes.clear()
 
         while len(self.__database_changes) > 0:
             self.__database_changes.popitem()[1].close()
@@ -529,7 +555,154 @@ def initialize(self) -> DocumentStore:
         self._initialized = True
         return self
 
-    # todo: aggressively cache
+    def aggressively_cache_for(
+        self,
+        cache_duration: datetime.timedelta,
+        database: Optional[str] = None,
+        mode: Optional["AggressiveCacheMode"] = None,
+    ) -> "DocumentStore._AggressiveCacheContext":
+        from ravendb.http.misc import AggressiveCacheMode
+
+        if mode is None:
+            mode = AggressiveCacheMode.TRACK_CHANGES
+        context = self._set_aggressive_cache(cache_duration, mode, database)
+        return self._finalize_aggressive_cache(context, mode, database)
+
+    def _set_aggressive_cache(
+        self,
+        cache_duration: datetime.timedelta,
+        mode: "AggressiveCacheMode",
+        database: Optional[str] = None,
+    ) -> "DocumentStore._AggressiveCacheContext":
+        from ravendb.http.misc import AggressiveCacheOptions
+
+        self.assert_initialized()
+        database = self.get_effective_database(database)
+        request_executor = self.get_request_executor(database)
+        options = AggressiveCacheOptions(cache_duration, mode)
+        return DocumentStore._AggressiveCacheContext(request_executor, options)
+
+    def _finalize_aggressive_cache(
+        self,
+        context: "DocumentStore._AggressiveCacheContext",
+        mode: "AggressiveCacheMode",
+        database: Optional[str] = None,
+    ) -> "DocumentStore._AggressiveCacheContext":
+        from ravendb.http.misc import AggressiveCacheMode
+
+        try:
+            if mode != AggressiveCacheMode.DO_NOT_TRACK_CHANGES:
+                database = self.get_effective_database(database)
+                self._listen_to_changes_and_update_cache(database)
+            return context
+        except Exception:
+            context.__enter__()
+            context.__exit__(None, None, None)
+            raise
+
+    def _listen_to_changes_and_update_cache(self, database: str) -> None:
+        if database in self.__aggressive_cache_changes:
+            return
+        # This lock achieves ConcurrentDict-like behavior
+        eviction = DocumentStore._AggressiveCacheInvalidator(self, database)
+        with self.__add_change_lock:
+            if database not in self.__aggressive_cache_changes:
+                self.__aggressive_cache_changes[database] = eviction
+                eviction.ensure_connected()
+            else:
+                eviction.close()
+
+    def disable_aggressive_caching(
+        self, database: Optional[str] = None
+    ) -> "DocumentStore._DisableAggressiveCachingContext":
+        self.assert_initialized()
+        database = self.get_effective_database(database)
+        request_executor = self.get_request_executor(database)
+        return DocumentStore._DisableAggressiveCachingContext(request_executor)
+
+    class _AggressiveCacheContext:
+        def __init__(self, request_executor, options):
+            self._request_executor = request_executor
+            self._options = options
+            self._old_options = None
+
+        def __enter__(self):
+            self._old_options = self._request_executor.aggressive_caching
+            self._request_executor.aggressive_caching = self._options
+            return self
+
+        def __exit__(self, exc_type, exc_val, exc_tb):
+            self._request_executor.aggressive_caching = self._old_options
+
+    class _DisableAggressiveCachingContext:
+        def __init__(self, request_executor):
+            self._request_executor = request_executor
+            self._old_options = None
+
+        def __enter__(self):
+            self._old_options = self._request_executor.aggressive_caching
+            self._request_executor.aggressive_caching = None
+            return self
+
+        def __exit__(self, exc_type, exc_val, exc_tb):
+            self._request_executor.aggressive_caching = self._old_options
+
+    class _AggressiveCacheInvalidator:
+        """Subscribes to document/index changes and invalidates the request executor's aggressive cache."""
+
+        def __init__(self, store: "DocumentStore", database: str):
+            from ravendb.changes.observers import ActionObserver
+            from ravendb.changes.types import DocumentChangeType, IndexChangeTypes
+
+            self._request_executor = store.get_request_executor(database)
+            self._changes = store.changes(database)
+            self._unsubscribers: List[Callable[[], None]] = []
+
+            # Capture by reference so the lambdas below always see the live cache object,
+            # even if RequestExecutor.cache is replaced. (It isn't today, but be explicit.)
+            cache_ref = self._request_executor.cache
+
+            def _invalidate() -> None:
+                cache_ref.generation += 1
+
+            def on_document_change(change) -> None:
+                # Only Put and Delete affect query results; ConflictResolved etc. do not.
+                if change.type_of_change in (DocumentChangeType.PUT, DocumentChangeType.DELETE):
+                    _invalidate()
+
+            def on_index_change(change) -> None:
+                # BatchCompleted means new index results are available; IndexRemoved means
+                # stale queries might have been using it.
+                if change.type_of_change in (IndexChangeTypes.BATCH_COMPLETED, IndexChangeTypes.INDEX_REMOVED):
+                    _invalidate()
+
+            # subscribe_with_observer (not subscribe) so we can attach an on_error callback.
+            # subscribe() creates an ActionObserver with no on_error, which means a WebSocket
+            # disconnect silently swallows the error — cache.generation is never bumped and
+            # the aggressive cache serves stale data indefinitely after the connection dies.
+            self._unsubscribers.append(
+                self._changes.for_all_documents().subscribe_with_observer(
+                    ActionObserver(on_next=on_document_change, on_error=lambda e: _invalidate())
+                )
+            )
+            self._unsubscribers.append(
+                self._changes.for_all_indexes().subscribe_with_observer(
+                    ActionObserver(on_next=on_index_change, on_error=lambda e: _invalidate())
+                )
+            )
+
+        def ensure_connected(self) -> None:
+            # todo: DatabaseChanges auto-connects on __init__; add explicit
+            # ensure_connected_now() for parity with C# EnsureConnectedNow()
+            pass
+
+        def close(self) -> None:
+            for unsub in self._unsubscribers:
+                try:
+                    unsub()
+                except Exception:
+                    pass
+            self._unsubscribers.clear()
 
     def bulk_insert(self, database_name: str = None, options: BulkInsertOptions = None) -> BulkInsertOperation:
         self.assert_initialized()
diff --git a/ravendb/http/http_cache.py b/ravendb/http/http_cache.py
@@ -41,6 +41,8 @@ def age(self) -> datetime.timedelta:
 
     @property
     def might_have_been_modified(self) -> bool:
+        if self.item is None:
+            return True
         return self.item.generation != self.__cache_generation
 
     def close(self):
@@ -73,7 +75,7 @@ def __setitem__(self, key, value):
         self.__items.__setitem__(key, value)
 
     def __getitem__(self, item):
-        self.__items.__getitem__(item)
+        return self.__items.__getitem__(item)
 
     def close(self):
         self.__items.clear()
@@ -108,29 +110,3 @@ def set_not_found(self, url: str, aggressively_cached: bool) -> None:
             {ItemFlags.AGGRESSIVELY_CACHED, ItemFlags.NOT_FOUND} if aggressively_cached else {ItemFlags.NOT_FOUND}
         )
         self.__items[url] = http_cache_item
-
-    class ReleaseCacheItem:
-        def __init__(self, item: HttpCacheItem = None):
-            self.item: Union[None, HttpCacheItem] = item
-            self.__cache_generation = item.cache.generation if item else 0
-
-        def __enter__(self):
-            return self
-
-        def __exit__(self, exc_type, exc_val, exc_tb):
-            pass
-
-        def not_modified(self) -> None:
-            if self.item is not None:
-                self.item.last_server_update = datetime.datetime.now()
-                self.item.generation = self.__cache_generation
-
-        @property
-        def age(self) -> datetime.timedelta:
-            if self.item is None:
-                return datetime.timedelta.max
-            return datetime.datetime.now() - self.item.last_server_update
-
-        @property
-        def might_have_been_modified(self) -> bool:
-            return self.item.generation != self.__cache_generation
diff --git a/ravendb/http/request_executor.py b/ravendb/http/request_executor.py
@@ -8,7 +8,7 @@
 from concurrent.futures import ThreadPoolExecutor, Future, FIRST_COMPLETED, wait, ALL_COMPLETED
 import uuid
 from json import JSONDecodeError
-from threading import Timer, Semaphore, Lock
+from threading import Timer, Semaphore, Lock, local
 
 import requests
 from copy import copy
@@ -27,8 +27,14 @@
 from ravendb.exceptions.raven_exceptions import ClientVersionMismatchException
 
 
-from ravendb.http.http_cache import HttpCache
-from ravendb.http.misc import ReadBalanceBehavior, ResponseDisposeHandling, LoadBalanceBehavior, Broadcast
+from ravendb.http.http_cache import HttpCache, ItemFlags, ReleaseCacheItem
+from ravendb.http.misc import (
+    ReadBalanceBehavior,
+    ResponseDisposeHandling,
+    LoadBalanceBehavior,
+    Broadcast,
+    AggressiveCacheMode,
+)
 from ravendb.http.raven_command import RavenCommand, RavenCommandResponseType
 from ravendb.http.server_node import ServerNode
 from ravendb.http.topology import Topology, NodeStatus, NodeSelector, CurrentIndexAndNode, UpdateTopologyParameters
@@ -106,13 +112,25 @@ def __init__(
         self._disposed: Union[None, bool] = None
 
         self.__synchronized_lock = Lock()
+        self._aggressive_caching_local = local()
 
         # --- events ---
         self._on_before_request: List[Callable[[BeforeRequestEventArgs], Any]] = []
         self.__on_failed_request: List[Callable[[FailedRequestEventArgs], None]] = []
         self.__on_succeed_request: List[Callable[[SucceedRequestEventArgs], None]] = []
         self._on_topology_updated: List[Callable[[Topology], None]] = []
 
+    @property
+    def aggressive_caching(self) -> Optional["AggressiveCacheOptions"]:
+        # threading.local mirrors C#'s AsyncLocal<AggressiveCacheOptions>: each thread
+        # (each request context) gets its own setting. Without this, one thread enabling
+        # aggressive caching would bleed into every other thread on the same executor.
+        return getattr(self._aggressive_caching_local, "value", None)
+
+    @aggressive_caching.setter
+    def aggressive_caching(self, value: Optional["AggressiveCacheOptions"]) -> None:
+        self._aggressive_caching_local.value = value
+
     def __enter__(self):
         return self
 
@@ -522,7 +540,38 @@ def execute(
         no_caching = session_info.no_caching if session_info else False
 
         cached_item, change_vector, cached_value = self._get_from_cache(command, not no_caching, url)
-        # todo: if change_vector exists try get from cache - aggressive caching
+
+        # Aggressive-cache short-circuit: serve from local cache without touching the server.
+        # All five conditions must hold:
+        #   1. Session didn't disable caching.
+        #   2. Aggressive caching is active on this thread.
+        #   3. The command doesn't opt out (streaming commands set can_cache_aggressively=False).
+        #   4. The item is actually in cache AND young enough.
+        #   5. Under TRACK_CHANGES mode, the cache generation must not have advanced since we
+        #      retrieved the item — if it has, _AggressiveCacheEviction saw a server change and
+        #      bumped the generation, so we must revalidate.
+        if (
+            not no_caching
+            and self.aggressive_caching is not None
+            and command.can_cache_aggressively
+            and cached_item.item is not None
+            and cached_item.age < self.aggressive_caching.duration
+            and (
+                not cached_item.might_have_been_modified
+                or self.aggressive_caching.mode != AggressiveCacheMode.TRACK_CHANGES
+            )
+        ):
+            if ItemFlags.NOT_FOUND in cached_item.item.flags:
+                # Cached 404: only trust it when it was itself received inside an aggressive-
+                # cache context (AGGRESSIVELY_CACHED flag set by set_not_found). A 404 cached
+                # outside aggressive mode might have been a transient error; re-fetch it.
+                if ItemFlags.AGGRESSIVELY_CACHED in cached_item.item.flags:
+                    command.set_response(None, True)
+                    return
+            elif cached_value is not None:
+                command.set_response(cached_value, True)
+                return
+
         with cached_item:
             # todo: try get from cache
             self._set_request_headers(session_info, change_vector, request)
@@ -785,7 +834,7 @@ def _set_request_headers(
 
     def _get_from_cache(
         self, command: RavenCommand, use_cache: bool, url: str
-    ) -> Tuple[HttpCache.ReleaseCacheItem, Optional[str], Optional[str]]:
+    ) -> Tuple[ReleaseCacheItem, Optional[str], Optional[str]]:
         if (
             use_cache
             and command.can_cache
@@ -794,7 +843,7 @@ def _get_from_cache(
         ):
             return self._cache.get(url)
 
-        return HttpCache.ReleaseCacheItem(), None, None
+        return ReleaseCacheItem(), None, None
 
     @staticmethod
     def __try_get_server_version(response: requests.Response) -> Union[None, str]:
@@ -1014,7 +1063,7 @@ def _handle_unsuccessful_response(
         should_retry: bool,
     ) -> bool:
         if response.status_code == HTTPStatus.NOT_FOUND:
-            self._cache.set_not_found(url, False)  # todo : check if aggressively cached, don't just pass False
+            self._cache.set_not_found(url, self.aggressive_caching is not None)
             if command.response_type == RavenCommandResponseType.EMPTY:
                 return True
             elif command.response_type == RavenCommandResponseType.OBJECT:
diff --git a/ravendb/tests/issue_tests/test_RDBC_1035.py b/ravendb/tests/issue_tests/test_RDBC_1035.py
