RDBC-1048 Fix silent data corruption bugs in Document Session

.github/pull_request_template.md

@@ -0,0 +1,60 @@
+### Issue link
+
+https://issues.hibernatingrhinos.com/issue/RDBC-...
+
+### Additional description
+
+...Include details of the change made in this Pull Request or additional notes for the solution. Anything that can be useful for reviewers of this PR...
+
+### Type of change
+
+- [ ] Bug fix
+- [ ] Regression bug fix
+- [ ] Optimization
+- [ ] New feature
+
+### How risky is the change?
+
+- [ ] Low
+- [ ] Moderate
+- [ ] High
+- [ ] Not relevant
+
+### Backward compatibility
+
+- [ ] Non breaking change
+- [ ] Ensured. Please explain how has it been implemented?
+- [ ] Breaking change
+- [ ] Not relevant
+
+### Is it platform specific issue?
+
+- [ ] Yes. Please list the affected platforms.
+- [ ] No
+
+### Documentation update
+
+- [ ] This change requires a documentation update. Please mark the issue on YouTrack using `Documentation Required` tag.
+- [ ] No documentation update is needed
+
+### Testing by Contributor
+
+- [ ] Tests have been added that prove the fix is effective or that the feature works
+- [ ] Internal classes added to the test class (e.g. entity or index definition classes) have the lowest possible access modifier (preferable `private`)
+- [ ] It has been verified by manual testing
+- [ ] Existing tests verify the correct behavior
+
+### Testing by RavenDB QA team
+
+- [ ] This change requires a special QA testing due to possible performance or resources usage implications (CPU, memory, IO). Please mark the issue on YouTrack using `QA Required` tag.
+- [ ] No special testing by RavenDB QA team is needed
+
+### Is there any existing behavior change of other features due to this change?
+
+- [ ] Yes. Please list the affected features/subsystems and provide appropriate explanation
+- [ ] No
+
+### UI work
+
+- [ ] It requires further work in the Studio. Please mark the issue on YouTrack using `Studio Required` tag.
+- [ ] No UI work is needed

ravendb/documents/operations/batch.py

@@ -127,9 +127,9 @@ def get_command_type(obj_node: dict) -> CommandType:
             elif command_type == CommandType.COUNTERS:
                 self._handle_counters(batch_result)
             elif command_type == CommandType.TIME_SERIES:
-                break  # todo: RavenDB-13474 add to time series cache
+                continue  # todo: RavenDB-13474 add to time series cache
             elif command_type == CommandType.TIME_SERIES_COPY or command_type == CommandType.BATCH_PATCH:
-                break
+                continue
             else:
                 raise ValueError(f"Command {command_type} is not supported")
 
@@ -190,7 +190,7 @@ def _handle_patch(self, batch_result: dict) -> None:
             self._throw_missing_field(CommandType.PATCH, "PatchStatus")
 
         status = PatchStatus(patch_status)
-        if status == PatchStatus.CREATED or PatchStatus.PATCHED:
+        if status in (PatchStatus.CREATED, PatchStatus.PATCHED):
             document = batch_result.get("ModifiedDocument")
             if not document:
                 return

ravendb/documents/operations/compare_exchange/compare_exchange.py

@@ -133,7 +133,7 @@ def delete(self, index: int) -> None:
         self._state = CompareExchangeValueState.DELETED
 
     def __assert_state(self) -> None:
-        if self._state == CompareExchangeValueState.NONE or CompareExchangeValueState.MISSING:
+        if self._state in (CompareExchangeValueState.NONE, CompareExchangeValueState.MISSING):
             return
         elif self._state == CompareExchangeValueState.CREATED:
             raise RuntimeError(f"The compare exchange value with key {self._key} was already stored.")
@@ -144,7 +144,7 @@ def get_command(
         self, conventions: DocumentConventions
     ) -> Optional[Union[DeleteCompareExchangeCommandData, PutCompareExchangeCommandData]]:
         s = self._state
-        if s == CompareExchangeValueState.NONE or CompareExchangeValueState.CREATED:
+        if s in (CompareExchangeValueState.NONE, CompareExchangeValueState.CREATED):
             if not self.__value:
                 return None
 

ravendb/documents/session/document_session.py

@@ -828,7 +828,8 @@ def wait_for_replication_after_save_changes(
             builder_options = builder.get_options()
             replication_options = builder_options.replication_options
             if replication_options is None:
-                builder_options.replication_options = ReplicationBatchOptions()
+                replication_options = ReplicationBatchOptions()
+                builder_options.replication_options = replication_options
 
             if replication_options.wait_for_replicas_timeout is None:
                 replication_options.wait_for_replicas_timeout = (

ravendb/documents/session/document_session_operations/in_memory_document_session_operations.py

@@ -985,13 +985,13 @@ def validate_cluster_transaction(self, result: SaveChangesData) -> None:
             )
 
         for command_data in result.session_commands:
-            if command_data.command_type == CommandType.PUT or CommandType.DELETE:
+            if command_data.command_type in (CommandType.PUT, CommandType.DELETE):
                 if command_data.change_vector is not None:
                     raise ValueError(
                         f"Optimistic concurrency for {command_data.key} "
                         f"is not supported when using a cluster transaction"
                     )
-            elif command_data.command_type == CommandType.COMPARE_EXCHANGE_DELETE or CommandType.COMPARE_EXCHANGE_PUT:
+            elif command_data.command_type in (CommandType.COMPARE_EXCHANGE_DELETE, CommandType.COMPARE_EXCHANGE_PUT):
                 pass
             else:
                 raise ValueError(f"The command '{command_data.command_type}' is not supported in a cluster session.")
@@ -1818,9 +1818,10 @@ def _get_operation_result(self, object_type: Type[_T], result: _T) -> _T:
             # todo: cast result on object_type
         raise TypeError(f"Unable to cast {result.__class__.__name__} to {object_type.__name__}")
 
-    # todo: implement method below
     def update_session_after_save_changes(self, result: BatchCommandResult):
         returned_transaction_index = result.transaction_index
+        if returned_transaction_index is not None:
+            self.session_info.last_cluster_transaction_index = returned_transaction_index
 
     def _process_query_parameters(
         self, object_type: type, index_name: str, collection_name: str, conventions: DocumentConventions
@@ -1857,7 +1858,7 @@ def get_options(self) -> BatchOptions:
         def with_timeout(
             self, timeout: datetime.timedelta
         ) -> InMemoryDocumentSessionOperations.ReplicationWaitOptsBuilder:
-            self.get_options().replication_options.wait_for_indexes_timeout = timeout
+            self.get_options().replication_options.wait_for_replicas_timeout = timeout
             return self
 
         def throw_on_timeout(self, should_throw: bool) -> InMemoryDocumentSessionOperations.ReplicationWaitOptsBuilder:
@@ -1890,11 +1891,11 @@ def with_timeout(self, timeout: datetime.timedelta) -> InMemoryDocumentSessionOp
             return self
 
         def throw_on_timeout(self, should_throw: bool) -> InMemoryDocumentSessionOperations.IndexesWaitOptsBuilder:
-            self.get_options().index_options.throw_on_timeout_in_wait_for_replicas = should_throw
+            self.get_options().index_options.throw_on_timeout_in_wait_for_indexes = should_throw
             return self
 
         def wait_for_indexes(self, *indexes: str) -> InMemoryDocumentSessionOperations.IndexesWaitOptsBuilder:
-            self.get_options().index_options.wait_for_indexes = indexes
+            self.get_options().index_options.wait_for_specific_indexes = indexes
             return self
 
     class SaveChangesData:

ravendb/tests/jvm_migrated_tests/cluster_tests/test_cluster_transaction.py

@@ -1,3 +1,7 @@
+import unittest
+
+from ravendb.documents.commands.batches import CommandType, CountersBatchCommandData
+from ravendb.documents.operations.counters import CounterOperation, CounterOperationType
 from ravendb.documents.session.misc import SessionOptions, TransactionMode
 from ravendb.infrastructure.entities import User
 from ravendb.tests.test_base import TestBase
@@ -74,3 +78,64 @@ def test_session_sequence(self):
             user1.age = 10
             session.store(user1, "users/1")
             session.save_changes()
+
+    def test_throw_on_unsupported_operations(self):
+        session_options = SessionOptions(
+            transaction_mode=TransactionMode.CLUSTER_WIDE,
+            disable_atomic_document_writes_in_cluster_wide_transaction=True,
+        )
+
+        with self.store.open_session(session_options=session_options) as session:
+            from ravendb.documents.session.document_session_operations.in_memory_document_session_operations import (
+                InMemoryDocumentSessionOperations,
+            )
+
+            counter_op = CounterOperation("likes", CounterOperationType.INCREMENT, 1)
+            counter_cmd = CountersBatchCommandData("docs/1", counter_op)
+
+            save_changes_data = InMemoryDocumentSessionOperations.SaveChangesData(session)
+            save_changes_data.session_commands.append(counter_cmd)
+
+            with self.assertRaises(ValueError) as ctx:
+                session.validate_cluster_transaction(save_changes_data)
+
+            self.assertIn("not supported", str(ctx.exception))
+
+    def test_compare_exchange_double_create_raises(self):
+        session_options = SessionOptions(
+            transaction_mode=TransactionMode.CLUSTER_WIDE,
+            disable_atomic_document_writes_in_cluster_wide_transaction=True,
+        )
+
+        with self.store.open_session(session_options=session_options) as session:
+            session.advanced.cluster_transaction.create_compare_exchange_value("users/emails/john", "john@doe.com")
+
+            with self.assertRaises(RuntimeError):
+                session.advanced.cluster_transaction.create_compare_exchange_value("users/emails/john", "other@doe.com")
+
+
+class TestClusterTransactionValidation(unittest.TestCase):
+    def test_cluster_tx_rejects_unsupported_command_types(self):
+        import inspect
+        from ravendb.documents.session.document_session_operations.in_memory_document_session_operations import (
+            InMemoryDocumentSessionOperations,
+        )
+
+        src = inspect.getsource(InMemoryDocumentSessionOperations.validate_cluster_transaction)
+        self.assertNotIn(
+            "== CommandType.PUT or CommandType.DELETE",
+            src,
+            "Cluster TX validation uses 'x == A or B' (always True). Must use 'x in (A, B)'.",
+        )
+
+    def test_compare_exchange_rejects_double_create(self):
+        from ravendb.documents.operations.compare_exchange.compare_exchange import (
+            CompareExchangeSessionValue,
+            CompareExchangeValueState,
+        )
+
+        value = CompareExchangeSessionValue.__new__(CompareExchangeSessionValue)
+        value._key = "test"
+        value._state = CompareExchangeValueState.CREATED
+        with self.assertRaises(RuntimeError):
+            value._CompareExchangeSessionValue__assert_state()

ravendb/tests/session_tests/test_advanced.py

@@ -3,6 +3,7 @@
 from ravendb.documents.operations.indexes import PutIndexesOperation
 from ravendb.tests.test_base import TestBase
 from ravendb.exceptions.exceptions import InvalidOperationException
+import datetime
 import unittest
 import pathlib
 import os
@@ -126,6 +127,74 @@ def test_try_delete_attachment_putted_in_the_same_session(self):
                 with self.assertRaises(InvalidOperationException):
                     session.advanced.attachments.delete("users/1-A", "my_text_file")
 
+    def test_wait_for_replication_timeout_propagates(self):
+        with self.store.open_session() as session:
+            session.store(User("Idan", 30), "users/1-A")
+            session.advanced.wait_for_replication_after_save_changes(
+                lambda opts: opts.with_timeout(datetime.timedelta(seconds=5))
+            )
+            batch_options = session._save_changes_options
+            self.assertIsNotNone(batch_options)
+            self.assertIsNotNone(batch_options.replication_options)
+            self.assertEqual(batch_options.replication_options.wait_for_replicas_timeout, datetime.timedelta(seconds=5))
+
+    def test_wait_for_indexes_throw_on_timeout_propagates(self):
+        with self.store.open_session() as session:
+            session.store(User("Idan", 30), "users/1-A")
+            session.advanced.wait_for_indexes_after_save_changes(lambda opts: opts.throw_on_timeout(False))
+            batch_options = session._save_changes_options
+            self.assertIsNotNone(batch_options)
+            self.assertIsNotNone(batch_options.index_options)
+            self.assertIs(batch_options.index_options.throw_on_timeout_in_wait_for_indexes, False)
+
+    def test_wait_for_indexes_specific_indexes_propagates(self):
+        with self.store.open_session() as session:
+            session.store(User("Idan", 30), "users/1-A")
+            session.advanced.wait_for_indexes_after_save_changes(lambda opts: opts.wait_for_indexes("MyIndex"))
+            batch_options = session._save_changes_options
+            self.assertIsNotNone(batch_options)
+            self.assertIsNotNone(batch_options.index_options)
+            self.assertIn("MyIndex", batch_options.index_options.wait_for_specific_indexes)
+
+
+class _FakeSession:
+    def __init__(self):
+        self._save_changes_options = None
+
+
+class TestWaitForOptions(unittest.TestCase):
+    def test_replication_timeout_propagates(self):
+        from ravendb.documents.session.document_session_operations.in_memory_document_session_operations import (
+            InMemoryDocumentSessionOperations,
+        )
+
+        session = _FakeSession()
+        builder = InMemoryDocumentSessionOperations.ReplicationWaitOptsBuilder(session)
+        builder.with_timeout(datetime.timedelta(seconds=5))
+        self.assertEqual(
+            session._save_changes_options.replication_options.wait_for_replicas_timeout, datetime.timedelta(seconds=5)
+        )
+
+    def test_indexes_throw_on_timeout_propagates(self):
+        from ravendb.documents.session.document_session_operations.in_memory_document_session_operations import (
+            InMemoryDocumentSessionOperations,
+        )
+
+        session = _FakeSession()
+        builder = InMemoryDocumentSessionOperations.IndexesWaitOptsBuilder(session)
+        builder.throw_on_timeout(False)
+        self.assertIs(session._save_changes_options.index_options.throw_on_timeout_in_wait_for_indexes, False)
+
+    def test_specific_indexes_propagates(self):
+        from ravendb.documents.session.document_session_operations.in_memory_document_session_operations import (
+            InMemoryDocumentSessionOperations,
+        )
+
+        session = _FakeSession()
+        builder = InMemoryDocumentSessionOperations.IndexesWaitOptsBuilder(session)
+        builder.wait_for_indexes("MyIndex")
+        self.assertIn("MyIndex", session._save_changes_options.index_options.wait_for_specific_indexes)
+
 
 if __name__ == "__main__":
     unittest.main()

ravendb/tests/session_tests/test_time_series.py

@@ -1,3 +1,5 @@
+import unittest
+
 from ravendb.tests.test_base import TestBase, User
 from datetime import datetime, timedelta
 
@@ -65,3 +67,40 @@ def test_time_series_cache(self):
 
             tsf.get(base + timedelta(days=2), base + timedelta(days=6))
             self.assertEqual(session.advanced.number_of_requests, 4)
+
+    def test_batch_processes_all_results_after_time_series(self):
+        with self.store.open_session() as session:
+            session.store(User("users/ts-target", "Target"))
+            session.save_changes()
+
+        with self.store.open_session() as session:
+            session.store(User("users/new-doc", "NewDoc"))
+            tsf = session.time_series_for("users/ts-target", "HeartRate")
+            tsf.append_single(datetime.now(), 70, "watch")
+            session.save_changes()
+
+        with self.store.open_session() as session:
+            new_doc = session.load("users/new-doc", User)
+            self.assertIsNotNone(new_doc, "PUT after time series operation must be processed.")
+            self.assertEqual(new_doc.name, "NewDoc")
+
+
+class TestBatchResultProcessing(unittest.TestCase):
+    def test_batch_does_not_break_after_time_series(self):
+        import inspect
+        from ravendb.documents.operations.batch import BatchOperation
+
+        src = inspect.getsource(BatchOperation)
+        lines = src.split("\n")
+        for i, line in enumerate(lines):
+            if "CommandType.TIME_SERIES" in line:
+                for j in range(i + 1, min(i + 3, len(lines))):
+                    next_line = lines[j].strip()
+                    if next_line and not next_line.startswith("#"):
+                        self.assertNotEqual(
+                            next_line,
+                            "break",
+                            "Batch processing uses 'break' after TIME_SERIES, "
+                            "dropping all subsequent results. Must use 'continue'.",
+                        )
+                        break