Merge branch 'master' into users/nalutripician/issue-5095-contacted-regions-hub-fallback

Author: NaluTripician

Directory.Build.props

@@ -1,7 +1,7 @@
 <Project xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
 	<PropertyGroup>
-		<ClientOfficialVersion>3.58.0</ClientOfficialVersion>
-		<ClientPreviewVersion>3.59.0</ClientPreviewVersion>
+		<ClientOfficialVersion>3.59.0</ClientOfficialVersion>
+		<ClientPreviewVersion>3.60.0</ClientPreviewVersion>
 		<ClientPreviewSuffixVersion>preview.0</ClientPreviewSuffixVersion>
 		<DirectVersion>3.42.4</DirectVersion>
 		<FaultInjectionVersion>1.0.0</FaultInjectionVersion>

Microsoft.Azure.Cosmos.Encryption.Custom/src/Microsoft.Azure.Cosmos.Encryption.Custom.csproj

@@ -68,6 +68,10 @@
   </PropertyGroup>
 
   <PropertyGroup Condition=" '$(SdkProjectRef)' == 'True' ">
+    <DefineSdkProjectRefSymbol Condition=" '$(DefineSdkProjectRefSymbol)' == '' ">true</DefineSdkProjectRefSymbol>
+  </PropertyGroup>
+
+  <PropertyGroup Condition=" '$(DefineSdkProjectRefSymbol)' == 'true' ">
     <DefineConstants>$(DefineConstants);SDKPROJECTREF</DefineConstants>
   </PropertyGroup>
 </Project>

Microsoft.Azure.Cosmos.Encryption/src/Microsoft.Azure.Cosmos.Encryption.csproj

@@ -74,6 +74,10 @@
   </PropertyGroup>
 
   <PropertyGroup Condition=" '$(SdkProjectRef)' == 'True' ">
+    <DefineSdkProjectRefSymbol Condition=" '$(DefineSdkProjectRefSymbol)' == '' ">true</DefineSdkProjectRefSymbol>
+  </PropertyGroup>
+
+  <PropertyGroup Condition=" '$(DefineSdkProjectRefSymbol)' == 'true' ">
     <DefineConstants>$(DefineConstants);SDKPROJECTREF</DefineConstants>
   </PropertyGroup>
 </Project>

Microsoft.Azure.Cosmos/contracts/API_3.59.0.txt

@@ -5,6 +5,7 @@
 namespace Microsoft.Azure.Cosmos
 {
     using System;
+    using System.IO;
     using Microsoft.Azure.Cosmos.ChangeFeed.Configuration;
     using Microsoft.Azure.Cosmos.ChangeFeed.LeaseManagement;
     using static Microsoft.Azure.Cosmos.Container;
@@ -219,22 +220,73 @@ public ChangeFeedProcessorBuilder WithLeaseContainer(Container leaseContainer)
         /// <returns>The instance of <see cref="ChangeFeedProcessorBuilder"/> to use.</returns>
         public virtual ChangeFeedProcessorBuilder WithInMemoryLeaseContainer()
         {
-            if (this.leaseContainer != null)
+            this.ValidateNoLeaseContainerConfigured();
+
+            if (string.IsNullOrEmpty(this.InstanceName))
             {
-                throw new InvalidOperationException("The builder already defined a lease container.");
+                this.InstanceName = ChangeFeedProcessorBuilder.InMemoryDefaultHostName;
             }
 
-            if (this.LeaseStoreManager != null)
+            this.LeaseStoreManager = new DocumentServiceLeaseStoreManagerInMemory();
+            return this;
+        }
+
+        /// <summary>
+        /// Uses an in-memory container to maintain state of the leases, optionally initialized from a <see cref="MemoryStream"/>
+        /// containing previously persisted lease state.
+        /// 
+        /// When the processor is stopped via <see cref="ChangeFeedProcessor.StopAsync"/>, the current lease state
+        /// is automatically written back to the same <paramref name="leaseState"/> stream, allowing the state to be
+        /// restored when creating a new processor instance.
+        /// 
+        /// Using an in-memory container restricts the scaling capability to just the instance running the current processor.
+        /// </summary>
+        /// <remarks>
+        /// <para>
+        /// <see cref="ChangeFeedProcessor.StopAsync"/> must not be invoked concurrently from multiple threads; the
+        /// in-memory container expects a single shutdown call and does not synchronize concurrent writers to
+        /// <paramref name="leaseState"/>.
+        /// </para>
+        /// </remarks>
+        /// <param name="leaseState">
+        /// A <see cref="MemoryStream"/> that serves as both input and output for lease state.
+        /// If the stream contains data, leases are deserialized and used to initialize the container.
+        /// When the processor stops, the current lease state is serialized back into this stream.
+        /// The stream must be writable and expandable (for example, created via <c>new MemoryStream()</c>).
+        /// A fixed-size stream such as <c>new MemoryStream(byte[])</c> will fail at shutdown if the
+        /// serialized lease state exceeds the original buffer capacity.
+        /// A <see cref="MemoryStream"/> is required (rather than the base <see cref="System.IO.Stream"/> type) so that
+        /// the lease state can be trimmed via <see cref="MemoryStream.SetLength(long)"/> when a new snapshot is smaller
+        /// than the previously persisted one. To integrate with <see cref="System.IO.Stream"/>-based persistence
+        /// (e.g., a file or blob), call <see cref="MemoryStream.ToArray"/> after <see cref="ChangeFeedProcessor.StopAsync"/>
+        /// to obtain the persisted bytes; create an expandable
+        /// <see cref="MemoryStream"/> (<c>new MemoryStream()</c>), write the bytes into it, set
+        /// <see cref="System.IO.Stream.Position"/> back to 0, and pass it to this method.
+        /// </param>
+        /// <returns>The instance of <see cref="ChangeFeedProcessorBuilder"/> to use.</returns>
+        /// <exception cref="ArgumentNullException">Thrown when <paramref name="leaseState"/> is null.</exception>
+        public virtual ChangeFeedProcessorBuilder WithInMemoryLeaseContainer(MemoryStream leaseState)
+        {
+            if (leaseState == null)
             {
-                throw new InvalidOperationException("The builder already defined an in-memory lease container instance.");
+                throw new ArgumentNullException(nameof(leaseState));
+            }
+
+            this.ValidateNoLeaseContainerConfigured();
+
+            if (!leaseState.CanWrite)
+            {
+                throw new ArgumentException("The lease state stream must be writable so that state can be persisted on shutdown.", nameof(leaseState));
             }
 
             if (string.IsNullOrEmpty(this.InstanceName))
             {
                 this.InstanceName = ChangeFeedProcessorBuilder.InMemoryDefaultHostName;
             }
 
-            this.LeaseStoreManager = new DocumentServiceLeaseStoreManagerInMemory();
+            // Deserialization of lease state (if any) is handled inside the manager
+            // so that serialization and deserialization are co-located in the same layer.
+            this.LeaseStoreManager = new DocumentServiceLeaseStoreManagerInMemory(leaseState);
             return this;
         }
 
@@ -317,5 +369,18 @@ public ChangeFeedProcessor Build()
             this.isBuilt = true;
             return this.changeFeedProcessor;
         }
+
+        private void ValidateNoLeaseContainerConfigured()
+        {
+            if (this.leaseContainer != null)
+            {
+                throw new InvalidOperationException("The builder already defined a lease container.");
+            }
+
+            if (this.LeaseStoreManager != null)
+            {
+                throw new InvalidOperationException("The builder already defined an in-memory lease container instance.");
+            }
+        }
     }
 }

Microsoft.Azure.Cosmos/contracts/API_3.60.0-preview.0.txt

@@ -5,7 +5,6 @@
 namespace Microsoft.Azure.Cosmos.ChangeFeed
 {
     using System;
-    using System.Collections.Generic;
     using System.Threading.Tasks;
     using Microsoft.Azure.Cosmos.ChangeFeed.Bootstrapping;
     using Microsoft.Azure.Cosmos.ChangeFeed.Configuration;
@@ -79,7 +78,14 @@ public override async Task StartAsync()
         public override async Task StopAsync()
         {
             DefaultTrace.TraceInformation("Stopping processor...");
+
+            // Persist in-memory lease state before stopping the partition manager so that
+            // a subsequent partition-manager shutdown failure cannot prevent recovery of the
+            // lease snapshot. No-op for Cosmos-backed leases.
+            await this.documentServiceLeaseStoreManager.ShutdownAsync().ConfigureAwait(false);
+
             await this.partitionManager.StopAsync().ConfigureAwait(false);
+
             DefaultTrace.TraceInformation("Processor stopped.");
         }
 

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/ChangeFeedProcessorBuilder.cs

@@ -1,21 +1,32 @@
-//------------------------------------------------------------
+//------------------------------------------------------------
 // Copyright (c) Microsoft Corporation.  All rights reserved.
 //------------------------------------------------------------
 
 namespace Microsoft.Azure.Cosmos.ChangeFeed.LeaseManagement
 {
+    using System;
     using System.Collections.Concurrent;
     using System.Collections.Generic;
+    using System.IO;
     using System.Linq;
     using System.Threading.Tasks;
 
     internal sealed class DocumentServiceLeaseContainerInMemory : DocumentServiceLeaseContainer
     {
         private readonly ConcurrentDictionary<string, DocumentServiceLease> container;
+        private readonly MemoryStream leaseStateStream;
 
         public DocumentServiceLeaseContainerInMemory(ConcurrentDictionary<string, DocumentServiceLease> container)
+            : this(container, leaseStateStream: null)
+        {
+        }
+
+        public DocumentServiceLeaseContainerInMemory(
+            ConcurrentDictionary<string, DocumentServiceLease> container,
+            MemoryStream leaseStateStream)
         {
             this.container = container;
+            this.leaseStateStream = leaseStateStream;
         }
 
         public override Task<IReadOnlyList<DocumentServiceLease>> GetAllLeasesAsync()
@@ -27,5 +38,46 @@ public override Task<IEnumerable<DocumentServiceLease>> GetOwnedLeasesAsync()
         {
             return Task.FromResult<IEnumerable<DocumentServiceLease>>(this.container.Values.AsEnumerable());
         }
+
+        /// <summary>
+        /// Persists the current in-memory lease state into the user-supplied <see cref="MemoryStream"/>.
+        /// </summary>
+        /// <remarks>
+        /// Must only be invoked from the single <c>ChangeFeedProcessor.StopAsync</c> call path;
+        /// concurrent invocation is not supported and may corrupt the stream.
+        /// </remarks>
+        /// <returns>A completed task once the stream has been populated, or a no-op if no stream was supplied.</returns>
+        public Task ShutdownAsync()
+        {
+            if (this.leaseStateStream == null)
+            {
+                return Task.CompletedTask;
+            }
+
+            byte[] serializedBytes = InMemoryLeaseJsonFormat.Serialize(this.container.Values.ToList());
+
+            // Resize the target stream BEFORE writing. If the stream is not expandable and
+            // cannot hold the new payload, SetLength throws NotSupportedException and the
+            // user's stream is left untouched (no partial-write corruption). If SetLength
+            // succeeds, the subsequent Write is guaranteed to fit.
+            try
+            {
+                this.leaseStateStream.SetLength(serializedBytes.Length);
+            }
+            catch (NotSupportedException ex)
+            {
+                throw new InvalidOperationException(
+                    "Failed to persist lease state because the MemoryStream is not expandable and the serialized "
+                    + "state exceeds its capacity. Use 'new MemoryStream()' or a MemoryStream with sufficient "
+                    + "capacity instead of 'new MemoryStream(byte[])' to create a resizable stream.",
+                    ex);
+            }
+
+            this.leaseStateStream.Position = 0;
+            this.leaseStateStream.Write(serializedBytes, 0, serializedBytes.Length);
+            this.leaseStateStream.Position = 0;
+
+            return Task.CompletedTask;
+        }
     }
 }

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/ChangeFeedProcessorCore.cs

@@ -4,6 +4,8 @@
 
 namespace Microsoft.Azure.Cosmos.ChangeFeed.LeaseManagement
 {
+    using System.Threading.Tasks;
+
     /// <summary>
     /// The DocumentServiceLeaseStoreManager defines a way to perform operations with <see cref="DocumentServiceLease"/>.
     /// </summary>
@@ -29,5 +31,13 @@ internal abstract class DocumentServiceLeaseStoreManager
         /// for particular monitoring collection and lease container prefix.
         /// </summary>
         public abstract DocumentServiceLeaseStore LeaseStore { get; }
+
+        /// <summary>
+        /// Called when the processor is stopping. Implementations may override to perform
+        /// cleanup or state persistence. The default implementation (for Cosmos-backed
+        /// lease stores) is a no-op. Exceptions thrown from this method propagate to the
+        /// caller of <see cref="ChangeFeedProcessor.StopAsync"/>.
+        /// </summary>
+        public abstract Task ShutdownAsync();
     }
 }

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/LeaseManagement/DocumentServiceLeaseContainerInMemory.cs

@@ -5,6 +5,7 @@
 namespace Microsoft.Azure.Cosmos.ChangeFeed.LeaseManagement
 {
     using System;
+    using System.Threading.Tasks;
     using Microsoft.Azure.Cosmos;
 
     /// <summary>
@@ -80,5 +81,10 @@ internal DocumentServiceLeaseStoreManagerCosmos(
         public override DocumentServiceLeaseCheckpointer LeaseCheckpointer => this.leaseCheckpointer;
 
         public override DocumentServiceLeaseContainer LeaseContainer => this.leaseContainer;
+
+        public override Task ShutdownAsync()
+        {
+            return Task.CompletedTask;
+        }
     }
 }