Session Token False Progress (#42393)

* add session token false progress

* update changelog

* Apply suggestions from code review

Co-authored-by: Copilot <[email protected]>

* add more tests

* react to comments

* fix analyze

---------

Co-authored-by: Copilot <[email protected]>

Author: tvaron3

sdk/cosmos/azure-cosmos/CHANGELOG.md

@@ -9,6 +9,7 @@
 #### Bugs Fixed
 
 #### Other Changes
+* Added session token false progress merge logic. See [42393](https://github.com/Azure/azure-sdk-for-python/pull/42393)
 
 ### 4.14.0b2 (2025-08-12)
 

sdk/cosmos/azure-cosmos/azure/cosmos/_constants.py

@@ -51,6 +51,8 @@ class _Constants:
     HS_MAX_ITEMS_CONFIG_DEFAULT: int = 1000
     MAX_ITEM_BUFFER_VS_CONFIG: str = "AZURE_COSMOS_MAX_ITEM_BUFFER_VECTOR_SEARCH"
     MAX_ITEM_BUFFER_VS_CONFIG_DEFAULT: int = 50000
+    SESSION_TOKEN_FALSE_PROGRESS_MERGE_CONFIG: str = "AZURE_COSMOS_SESSION_TOKEN_FALSE_PROGRESS_MERGE"
+    SESSION_TOKEN_FALSE_PROGRESS_MERGE_CONFIG_DEFAULT: str = "True"
     CIRCUIT_BREAKER_ENABLED_CONFIG: str =  "AZURE_COSMOS_ENABLE_CIRCUIT_BREAKER"
     CIRCUIT_BREAKER_ENABLED_CONFIG_DEFAULT: str = "False"
     AAD_SCOPE_OVERRIDE: str = "AZURE_COSMOS_AAD_SCOPE_OVERRIDE"

sdk/cosmos/azure-cosmos/azure/cosmos/_vector_session_token.py

@@ -21,8 +21,10 @@
 
 """Session Consistency Tracking in the Azure Cosmos database service.
 """
+import os
 
 from . import exceptions
+from ._constants import _Constants
 from .http_constants import StatusCodes as _StatusCodes
 
 
@@ -113,9 +115,12 @@ def equals(self, other):
             and self.are_region_progress_equal(other.local_lsn_by_region)
         )
 
-    def merge(self, other):
+    def merge(self, other: "VectorSessionToken"):
         if other is None:
             raise ValueError("Invalid Session Token (should not be None)")
+        false_progress_merge_enabled = (os.environ.get(_Constants.SESSION_TOKEN_FALSE_PROGRESS_MERGE_CONFIG,
+                                                       _Constants.SESSION_TOKEN_FALSE_PROGRESS_MERGE_CONFIG_DEFAULT)
+                                        .lower() == "true")
 
         if self.version == other.version and len(self.local_lsn_by_region) != len(other.local_lsn_by_region):
             raise exceptions.CosmosHttpResponseError(
@@ -152,9 +157,12 @@ def merge(self, other):
                 )
             else:
                 highest_local_lsn_by_region[region_id] = local_lsn1
+        global_lsn = max(self.global_lsn, other.global_lsn)
+        if false_progress_merge_enabled and self.version != other.version:
+            global_lsn = session_token_with_higher_version.global_lsn
 
         return VectorSessionToken(
-            max(self.version, other.version), max(self.global_lsn, other.global_lsn), highest_local_lsn_by_region
+            max(self.version, other.version), global_lsn, highest_local_lsn_by_region
         )
 
     def convert_to_string(self):

sdk/cosmos/azure-cosmos/tests/test_session_token_unit.py

@@ -2,6 +2,7 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 
 import unittest
+import os
 
 import pytest
 
@@ -38,7 +39,7 @@ def test_validate_session_token_parsing_from_empty_string(self):
         session_token = ""
         self.assertIsNone(VectorSessionToken.create(session_token))
 
-    def test_validate_session_token_comparison(self):
+    def _different_merge_scenarios(self):
         # valid session token
         session_token1 = VectorSessionToken.create("1#100#1=20#2=5#3=30")
         session_token2 = VectorSessionToken.create("2#105#4=10#2=5#3=30")
@@ -73,11 +74,29 @@ def test_validate_session_token_comparison(self):
         self.assertIsNotNone(session_token_merged)
         self.assertTrue(session_token_merged.equals(session_token1.merge(session_token2)))
 
+        # same vector clock version with global lsn increase (no failover)
+        session_token1 = VectorSessionToken.create("1#100#1=20#2=5")
+        session_token2 = VectorSessionToken.create("1#197#1=20#2=5")
+        self.assertIsNotNone(session_token1)
+        self.assertIsNotNone(session_token2)
+
+        self.assertTrue(session_token1.merge(session_token2).equals(
+            VectorSessionToken.create("1#197#1=20#2=5")))
+
+        # same vector clock version with global lsn increase and local lsn increase
+        session_token1 = VectorSessionToken.create("1#100#1=20#2=5")
+        session_token2 = VectorSessionToken.create("1#197#1=23#2=15")
+        self.assertIsNotNone(session_token1)
+        self.assertIsNotNone(session_token2)
+
+        self.assertTrue(session_token1.merge(session_token2).equals(
+            VectorSessionToken.create("1#197#1=23#2=15")))
+
+        # different number of regions with same region should throw error
         session_token1 = VectorSessionToken.create("1#101#1=20#2=5#3=30")
         session_token2 = VectorSessionToken.create("1#100#1=20#2=5#3=30#4=40")
         self.assertIsNotNone(session_token1)
         self.assertIsNotNone(session_token2)
-
         try:
             session_token1.merge(session_token2)
             self.fail("Region progress can not be different when version is same")
@@ -86,6 +105,69 @@ def test_validate_session_token_comparison(self):
                              "Status code: 500\nCompared session tokens '1#101#1=20#2=5#3=30' "
                              "and '1#100#1=20#2=5#3=30#4=40' have unexpected regions.")
 
+        # same version with different region progress should throw error
+        session_token1 = VectorSessionToken.create("1#101#1=20#2=5#3=30")
+        session_token2 = VectorSessionToken.create("1#100#4=20#2=5#3=30")
+        self.assertIsNotNone(session_token1)
+        self.assertIsNotNone(session_token2)
+
+        try:
+            session_token1.merge(session_token2)
+            self.fail("Region progress can not be different when version is same")
+        except CosmosHttpResponseError as e:
+            self.assertEqual(str(e),
+                             "Status code: 500\nCompared session tokens '1#101#1=20#2=5#3=30' "
+                             "and '1#100#4=20#2=5#3=30' have unexpected regions.")
+
+    def test_validate_session_token_comparison(self):
+        self._different_merge_scenarios()
+        os.environ["AZURE_COSMOS_SESSION_TOKEN_FALSE_PROGRESS_MERGE"] = "false"
+        self._different_merge_scenarios()
+        del os.environ["AZURE_COSMOS_SESSION_TOKEN_FALSE_PROGRESS_MERGE"]
+
+    def test_session_token_false_progress_merge(self):
+        for false_progress_enabled in [True, False]:
+            self.validate_different_session_token_false_progress_merge_scenarios(false_progress_enabled)
+
+    def validate_different_session_token_false_progress_merge_scenarios(self, false_progress_enabled: bool):
+        # Test that false progress merge is enabled by default and that global lsn is used from higher version token
+        # when enabled
+        os.environ["AZURE_COSMOS_SESSION_TOKEN_FALSE_PROGRESS_MERGE"] = str(false_progress_enabled)
+        session_token1 = VectorSessionToken.create("1#200#1=20#2=5#3=30")
+        session_token2 = VectorSessionToken.create("2#100#1=10#2=8#3=30")
+        self.assertIsNotNone(session_token1)
+        self.assertIsNotNone(session_token2)
+        if false_progress_enabled:
+            expected_session_token = "2#100#1=20#2=8#3=30"
+        else:
+            expected_session_token = "2#200#1=20#2=8#3=30"
+        self.assertTrue(session_token1.merge(session_token2).equals(
+            VectorSessionToken.create(expected_session_token)))
+
+        # vector clock version increase with removed region progress should merge
+        session_token1 = VectorSessionToken.create("1#200#1=20#2=5#3=30")
+        session_token2 = VectorSessionToken.create("2#100#1=10#2=5")
+        self.assertIsNotNone(session_token1)
+        self.assertIsNotNone(session_token2)
+
+        if false_progress_enabled:
+            expected_session_token = "2#100#1=20#2=5"
+        else:
+            expected_session_token = "2#200#1=20#2=5"
+        self.assertTrue(session_token1.merge(session_token2).equals(
+            VectorSessionToken.create(expected_session_token)))
+
+        # vector clock version increase with new region progress should merge
+        session_token1 = VectorSessionToken.create("1#200#1=20#2=5")
+        session_token2 = VectorSessionToken.create("2#100#1=10#2=5#3=30")
+        self.assertIsNotNone(session_token1)
+        self.assertIsNotNone(session_token2)
+        if false_progress_enabled:
+            expected_session_token = "2#100#1=20#2=5#3=30"
+        else:
+            expected_session_token = "2#200#1=20#2=5#3=30"
+        self.assertTrue(session_token1.merge(session_token2).equals(
+            VectorSessionToken.create(expected_session_token)))
 
 if __name__ == '__main__':
     unittest.main()