RDBC-1059 Add OptimisticConcurrencyMode with WritesAndReads tracking

Author: poissoncorp

ravendb/__init__.py

@@ -239,6 +239,7 @@
     DocumentsChanges,
     ForceRevisionStrategy,
     MethodCall,
+    OptimisticConcurrencyMode,
     OrderingType,
     JavaScriptMap,
     DocumentQueryCustomization,

ravendb/documents/commands/batches.py

@@ -4,7 +4,7 @@
 import json
 from abc import abstractmethod
 from enum import Enum
-from typing import Callable, Union, Optional, TYPE_CHECKING, List, Set, Dict
+from typing import Callable, IO, Union, Optional, TYPE_CHECKING, List, Set, Dict
 
 import requests
 
@@ -49,6 +49,7 @@ class CommandType(Enum):
     TIME_SERIES_BULK_INSERT = "TIME_SERIES_BULK_INSERT"
     TIME_SERIES_COPY = "TIME_SERIES_COPY"
     BATCH_PATCH = "BatchPATCH"
+    BATCH_TRACK_CHANGES = "BatchTrackChanges"
     CLIENT_ANY_COMMAND = "CLIENT_ANY_COMMAND"
     CLIENT_MODIFY_DOCUMENT_COMMAND = "CLIENT_MODIFY_DOCUMENT_COMMAND"
 
@@ -81,6 +82,8 @@ def from_csharp_value_str(cls, value: str) -> CommandType:
             return cls.COUNTERS
         elif value == "BatchPATCH":
             return cls.BATCH_PATCH
+        elif value == "BatchTrackChanges":
+            return cls.BATCH_TRACK_CHANGES
         elif value == "ForceRevisionCreation":
             return cls.FORCE_REVISION_CREATION
         elif value == "TimeSeries":
@@ -119,7 +122,7 @@ def __init__(
                     self.__attachment_streams = []
                 stream = command.stream
                 if stream is None:
-                    continue  # remote-only attachment — no stream to track
+                    continue  # remote-only attachment has no local stream
                 if stream in self.__attachment_streams:
                     raise RuntimeError(
                         "It is forbidden to re-use the same stream for more than one attachment. "
@@ -270,6 +273,27 @@ def serialize(self, conventions: DocumentConventions) -> dict:
         pass
 
 
+class BatchTrackChangesCommandData(CommandData):
+    # Emitted in OptimisticConcurrencyMode.WRITES_AND_READS: carries the change
+    # vector of every tracked entity not already covered by a PUT/DELETE, so
+    # the server can verify none of them changed underneath us.
+    def __init__(self, tracked_entities: Dict[str, str], ids_to_skip: Set[str]):
+        super().__init__(command_type=CommandType.BATCH_TRACK_CHANGES)
+        self.tracked_entities = tracked_entities
+        self._ids_to_skip = ids_to_skip
+
+    def serialize(self, conventions: DocumentConventions) -> dict:
+        tracked = {
+            entity_id: change_vector
+            for entity_id, change_vector in self.tracked_entities.items()
+            if entity_id not in self._ids_to_skip
+        }
+        return {
+            "Type": str(CommandType.BATCH_TRACK_CHANGES),
+            "TrackedEntities": tracked,
+        }
+
+
 class DeleteCommandData(CommandData):
     def __init__(self, key: str, change_vector: str, original_change_vector: str = None):
         super(DeleteCommandData, self).__init__(key=key, command_type=CommandType.DELETE, change_vector=change_vector)
@@ -534,7 +558,7 @@ def __init__(
         self,
         document_id: str,
         name: str,
-        stream: bytes,
+        stream: Union[bytes, IO[bytes]],
         content_type: str,
         change_vector: str,
         remote_parameters: Optional["RemoteAttachmentParameters"] = None,
@@ -554,11 +578,11 @@ def __init__(
         self.__size_in_bytes = size_in_bytes
 
     @property
-    def stream(self):
+    def stream(self) -> Union[bytes, IO[bytes]]:
         return self.__stream
 
     @property
-    def content_type(self):
+    def content_type(self) -> str:
         return self.__content_type
 
     @property

ravendb/documents/conventions.py

@@ -51,7 +51,10 @@ def __init__(self):
 
         # Flags
         self.disable_topology_updates = False
-        self.use_optimistic_concurrency = False
+        self._optimistic_concurrency_mode = None
+        # Track which setter the user touched so we can reject mixing them.
+        self._use_optimistic_concurrency_was_set = False
+        self._optimistic_concurrency_mode_was_set = False
         self.throw_if_query_page_size_is_not_set = False
         self._send_application_identifier = True
         self._save_enums_as_integers: Optional[bool] = None
@@ -376,6 +379,39 @@ def _assert_not_frozen(self) -> None:
                 "Conventions has been frozen after documentStore.initialize()" " and no changes can be applied to them"
             )
 
+    @property
+    def optimistic_concurrency_mode(self):
+        from ravendb.documents.session.misc import OptimisticConcurrencyMode
+
+        return self._optimistic_concurrency_mode or OptimisticConcurrencyMode.NONE
+
+    @optimistic_concurrency_mode.setter
+    def optimistic_concurrency_mode(self, value) -> None:
+        self._assert_not_frozen()
+        if self._use_optimistic_concurrency_was_set:
+            raise RuntimeError("optimistic_concurrency_mode cannot be combined with use_optimistic_concurrency.")
+        self._optimistic_concurrency_mode_was_set = True
+        self._optimistic_concurrency_mode = value
+
+    @property
+    def use_optimistic_concurrency(self) -> bool:
+        from ravendb.documents.session.misc import OptimisticConcurrencyMode
+
+        return self._optimistic_concurrency_mode not in (None, OptimisticConcurrencyMode.NONE)
+
+    @use_optimistic_concurrency.setter
+    def use_optimistic_concurrency(self, value: bool) -> None:
+        # Legacy bool view: True <-> WRITES, False <-> NONE.
+        from ravendb.documents.session.misc import OptimisticConcurrencyMode
+
+        self._assert_not_frozen()
+        if self._optimistic_concurrency_mode_was_set:
+            raise RuntimeError("use_optimistic_concurrency cannot be combined with optimistic_concurrency_mode.")
+        self._use_optimistic_concurrency_was_set = True
+        self._optimistic_concurrency_mode = (
+            OptimisticConcurrencyMode.WRITES if value else OptimisticConcurrencyMode.NONE
+        )
+
     def clone(self) -> DocumentConventions:
         cloned = DocumentConventions()
         cloned._list_of_registered_id_conventions = [*self._list_of_registered_id_conventions]
@@ -392,7 +428,9 @@ def clone(self) -> DocumentConventions:
         cloned._find_collection_name = self._find_collection_name
         cloned._find_python_class_name = self.find_python_class_name
 
-        cloned.use_optimistic_concurrency = self.use_optimistic_concurrency
+        cloned._optimistic_concurrency_mode = self._optimistic_concurrency_mode
+        cloned._use_optimistic_concurrency_was_set = self._use_optimistic_concurrency_was_set
+        cloned._optimistic_concurrency_mode_was_set = self._optimistic_concurrency_mode_was_set
         cloned.throw_if_query_page_size_is_not_set = self.throw_if_query_page_size_is_not_set
         cloned.max_number_of_requests_per_session = self.max_number_of_requests_per_session
 

ravendb/documents/operations/batch.py

@@ -78,6 +78,7 @@ def get_command_type(obj_node: dict) -> CommandType:
                     "it. So it was executed ONLY on the requested node on " + self._session.request_executor.url
                 )
 
+        skip = 0
         for i in range(self._session_commands_count):
             batch_result = result.results[i]
             if batch_result is None:
@@ -86,7 +87,7 @@ def get_command_type(obj_node: dict) -> CommandType:
             command_type = get_command_type(batch_result)
 
             if command_type == CommandType.PUT:
-                self._handle_put(i, batch_result, False)
+                self._handle_put(i - skip, batch_result, False)
             elif command_type == CommandType.FORCE_REVISION_CREATION:
                 self._handle_force_revision_creation(batch_result)
             elif command_type == CommandType.DELETE:
@@ -95,6 +96,10 @@ def get_command_type(obj_node: dict) -> CommandType:
                 self._handle_compare_exchange_put(batch_result)
             elif command_type == CommandType.COMPARE_EXCHANGE_DELETE:
                 self._handle_compare_exchange_delete(batch_result)
+            elif command_type == CommandType.BATCH_TRACK_CHANGES:
+                # No client-side state to update; bump skip so PUT indices
+                # remain aligned with the SaveChangesData.entities array.
+                skip += 1
             else:
                 raise ValueError(f"Command {command_type} is not supported")
 
@@ -130,6 +135,8 @@ def get_command_type(obj_node: dict) -> CommandType:
                 continue  # todo: RavenDB-13474 add to time series cache
             elif command_type == CommandType.TIME_SERIES_COPY or command_type == CommandType.BATCH_PATCH:
                 continue
+            elif command_type == CommandType.BATCH_TRACK_CHANGES:
+                continue
             else:
                 raise ValueError(f"Command {command_type} is not supported")
 
@@ -229,6 +236,7 @@ def _handle_delete_internal(self, batch_result: dict, command_type: CommandType)
             return
 
         self._session.documents_by_id.pop(key, None)
+        self._session._tracked_entities.try_remove(key)
 
         if document_info.entity is not None:
             self._session.documents_by_entity.pop(document_info.entity, None)
@@ -306,6 +314,8 @@ def _handle_metadata_modifications(
         document_info.key = key
         document_info.change_vector = change_vector
 
+        self._session._tracked_entities.try_update(key, change_vector)
+
         self._apply_metadata_modifications(key, document_info)
 
     def _handle_counters(self, batch_result: Dict) -> None:

ravendb/documents/session/document_session.py

@@ -667,13 +667,24 @@ def transaction_mode(self) -> TransactionMode:
         def transaction_mode(self, value: TransactionMode):
             self._session.transaction_mode = value
 
+        @property
+        def optimistic_concurrency_mode(self):
+            return self._session._optimistic_concurrency_mode
+
+        @optimistic_concurrency_mode.setter
+        def optimistic_concurrency_mode(self, value):
+            self._session._set_optimistic_concurrency_mode(value)
+
         @property
         def use_optimistic_concurrency(self) -> bool:
-            return self._session._use_optimistic_concurrency
+            # Derived view; setter routes through optimistic_concurrency_mode.
+            from ravendb.documents.session.misc import OptimisticConcurrencyMode
+
+            return self._session._optimistic_concurrency_mode not in (None, OptimisticConcurrencyMode.NONE)
 
         @use_optimistic_concurrency.setter
         def use_optimistic_concurrency(self, value: bool):
-            self._session._use_optimistic_concurrency = value
+            self._session._set_use_optimistic_concurrency(value)
 
         def is_loaded(self, key: str) -> bool:
             return self._session.is_loaded_or_deleted(key)
@@ -753,6 +764,7 @@ def evict(self, entity: object) -> None:
                     self._session._counters_by_doc_id.pop(document_info.key, None)
                 if self._session.time_series_by_doc_id:
                     self._session.time_series_by_doc_id.pop(document_info.key, None)
+                self._session._tracked_entities.try_remove(document_info.key)
 
             self._session._deleted_entities.evict(entity)
             self._session.entity_to_json.remove_from_missing(entity)
@@ -772,6 +784,7 @@ def clear(self) -> None:
             self._session._clear_cluster_session()
             self._session._pending_lazy_operations.clear()
             self._session.entity_to_json.clear()
+            self._session._tracked_entities.clear()
 
         def document_query(
             self,