Merge branch 'main' into badrishc/native-allocator

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: a49f1db0-fd39-48c7-80dc-ce104c863b79

Author: badrishc

.github/workflows/ci.yml

@@ -159,7 +159,7 @@ jobs:
         os: [ ubuntu-latest, windows-latest ]
         framework: [ 'net8.0' , 'net10.0']
         configuration: [ 'Debug', 'Release' ]
-        test: [ 'Garnet.test.cluster', 'Garnet.test.cluster.migrate', 'Garnet.test.cluster.migrate.rangeindex', 'Garnet.test.cluster.replication', 'Garnet.test.cluster.replication.tls', 'Garnet.test.cluster.replication.disklesssync', 'Garnet.test.cluster.replication.rangeindex', 'Garnet.test.cluster.vectorsets', 'Garnet.test.cluster.multilog' ]
+        test: [ 'Garnet.test.cluster', 'Garnet.test.cluster.migrate', 'Garnet.test.cluster.migrate.rangeindex', 'Garnet.test.cluster.replication', 'Garnet.test.cluster.replication.tls', 'Garnet.test.cluster.replication.disklesssync', 'Garnet.test.cluster.replication.rangeindex', 'Garnet.test.cluster.replication.vectorsets', 'Garnet.test.cluster.vectorsets', 'Garnet.test.cluster.multilog' ]
     if: needs.changes.outputs.garnet == 'true'
     steps:
       - name: Check out code

.github/workflows/nightly.yml

@@ -51,7 +51,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        test: [ 'Garnet.test.cluster', 'Garnet.test.cluster.migrate', 'Garnet.test.cluster.replication', 'Garnet.test.cluster.replication.tls', 'Garnet.test.cluster.replication.asyncreplay', 'Garnet.test.cluster.replication.disklesssync', 'Garnet.test.cluster.vectorsets', 'Garnet.test.cluster.multilog' ]
+        test: [ 'Garnet.test.cluster', 'Garnet.test.cluster.migrate', 'Garnet.test.cluster.replication', 'Garnet.test.cluster.replication.tls', 'Garnet.test.cluster.replication.asyncreplay', 'Garnet.test.cluster.replication.disklesssync', 'Garnet.test.cluster.replication.vectorsets', 'Garnet.test.cluster.vectorsets', 'Garnet.test.cluster.multilog' ]
         os: [ ubuntu-latest, windows-latest ]
         framework: [ 'net8.0', 'net10.0' ]
         configuration: [ 'Debug', 'Release' ]

Garnet.slnx

@@ -98,5 +98,6 @@
     <Project Path="test/cluster/Garnet.test.cluster.replication.tls/Garnet.test.cluster.replication.tls.csproj" />
     <Project Path="test/cluster/Garnet.test.cluster.vectorsets/Garnet.test.cluster.vectorsets.csproj" />
     <Project Path="test/cluster/Garnet.test.cluster.replication.rangeindex/Garnet.test.cluster.replication.rangeindex.csproj" />
+    <Project Path="test/cluster/Garnet.test.cluster.replication.vectorsets/Garnet.test.cluster.replication.vectorsets.csproj" />
   </Folder>
 </Solution>

Version.props

@@ -1,6 +1,6 @@
 <Project>
 	<!-- VersionPrefix property for builds and packages -->
 	<PropertyGroup>
-		<VersionPrefix>2.1.2</VersionPrefix>
+		<VersionPrefix>2.1.3</VersionPrefix>
 	</PropertyGroup>
 </Project>

libs/cluster/Server/Replication/ReplicaOps/ReplicaDisklessSync.cs

@@ -234,6 +234,13 @@ public AofAddress TryReplicaDisklessRecovery(SyncMetadata primarySyncMetadata, o
                 // Before advertising updated replication offset, wait for Vector Set ops to finish
                 storeWrapper.DefaultDatabase.VectorManager?.WaitForVectorOperationsToComplete();
 
+                // Full sync only: a partial sync replays AOF instead of streaming raw records.
+                if (primarySyncMetadata.fullSync)
+                {
+                    // In full sync, a flush is done right before, so no reserved contexts should exist
+                    storeWrapper.DefaultDatabase.VectorManager?.ReconcileRecoveredState(requireNoReservedContexts: true);
+                }
+
                 this.replicationOffset = _replicationOffset;
 
                 // Mark this txn run as a read-write session if we are replaying as a replica

libs/cluster/Session/RespClusterReplicationCommands.cs

@@ -548,6 +548,7 @@ private bool NetworkClusterSync(out bool invalidParameters)
 
             TrackImportProgress(recordCount, recordCount == 0);
             var storeWrapper = clusterProvider.storeWrapper;
+            var vectorManager = storeWrapper.DefaultDatabase.VectorManager;
             var transientObjectIdMap = storeWrapper.store.Log.TransientObjectIdMap;
 
             DiskLogRecord diskLogRecord = default;
@@ -565,6 +566,10 @@ private bool NetworkClusterSync(out bool invalidParameters)
                             return false;
 
                         diskLogRecord = DiskLogRecord.Deserialize(recordSpan, storeWrapper.GarnetObjectSerializer, transientObjectIdMap, storeWrapper.storeFunctions);
+
+                        // Streamed records carry the primary's native handle and bypass the RMW path that maintains the context reservation
+                        vectorManager?.SanitizeAndTrackIngestedRecordIfApplicable(ref diskLogRecord);
+
                         _ = basicGarnetApi.SET(in diskLogRecord);
                         storeWrapper.storeFunctions.OnDisposeDiskRecord(ref diskLogRecord, DisposeReason.DeserializedFromDisk);
                         diskLogRecord.Dispose();

libs/common/Testing/ExceptionInjectionHelper.cs

@@ -72,6 +72,26 @@ public static void DisableException(ExceptionInjectionType exceptionType)
             _ = release.TrySetResult(true);
         }
 
+        /// <summary>
+        /// Enables an exception scenario until the returned scope is disposed.
+        /// </summary>
+        /// <param name="exceptionType"></param>
+        public static Scope EnabledScope(ExceptionInjectionType exceptionType)
+        {
+            EnableException(exceptionType);
+            return new Scope(exceptionType);
+        }
+
+        /// <summary>
+        /// Disables the exception scenario it was created for when disposed.
+        /// </summary>
+        /// <param name="exceptionType"></param>
+        public readonly struct Scope(ExceptionInjectionType exceptionType) : IDisposable
+        {
+            /// <inheritdoc/>
+            public void Dispose() => DisableException(exceptionType);
+        }
+
         /// <summary>
         /// Trigger exception scenario (NOTE: add this to the location where the exception should be emulated/triggered)
         /// </summary>

libs/server/Databases/MultiDatabaseManager.cs

@@ -1064,7 +1064,7 @@ public override void RecoverVectorSets()
             {
                 var dbId = activeDbIdsMapSnapshot[i];
                 databasesMapSnapshot[dbId].VectorManager.Initialize();
-                databasesMapSnapshot[dbId].VectorManager.ResumePostRecovery();
+                databasesMapSnapshot[dbId].VectorManager.ReconcileRecoveredState();
             }
         }
 

libs/server/Databases/SingleDatabaseManager.cs

@@ -401,7 +401,7 @@ public override void RecoverVectorSets()
             // Guarantee initialize has happened before we attempt to recover
             defaultDatabase.VectorManager.Initialize();
 
-            defaultDatabase.VectorManager.ResumePostRecovery();
+            defaultDatabase.VectorManager.ReconcileRecoveredState();
         }
 
         public override void Dispose()

libs/server/Properties/AssemblyInfo.cs

@@ -5,6 +5,7 @@
 
 [assembly: InternalsVisibleTo("Garnet.test" + AssemblyRef.GarnetPublicKey)]
 [assembly: InternalsVisibleTo("Garnet.test.cluster" + AssemblyRef.GarnetPublicKey)]
+[assembly: InternalsVisibleTo("Garnet.test.cluster.replication.vectorsets" + AssemblyRef.GarnetPublicKey)]
 [assembly: InternalsVisibleTo("Garnet.test.collections" + AssemblyRef.GarnetPublicKey)]
 [assembly: InternalsVisibleTo("Garnet.test.acl" + AssemblyRef.GarnetPublicKey)]
 [assembly: InternalsVisibleTo("Garnet.test.scripting" + AssemblyRef.GarnetPublicKey)]